InMemoryLKVLiveMarketDataProvider extends AbstractMarketDataProvider implements LiveMarketDataProvider, LiveDataListener, SubscriptionReporter { void resubscribe(Set<ExternalScheme> schemes) { _subscriptionReadLock.lock(); try { Collection<LiveDataSpecification> toSubscribe = new HashSet<>(); for (LiveDataSpecification requestedSpec : _pendingSubscriptionsByRequestedSpec.keySet()) { for (ExternalId id : requestedSpec.getIdentifiers()) { if (schemes.contains(id.getScheme())) { toSubscribe.add(requestedSpec); } } } for (LiveDataSpecification fullyQualifiedSpec : _activeSubscriptionsByQualifiedSpec.keySet()) { for (ExternalId id : fullyQualifiedSpec.getIdentifiers()) { if (schemes.contains(id.getScheme())) { toSubscribe.add(fullyQualifiedSpec); } } } if (!toSubscribe.isEmpty()) { s_logger.info("Subscribing {} to {} live data specifications", _marketDataUser, toSubscribe.size()); _liveDataClient.subscribe(_marketDataUser, toSubscribe, this); } } finally { _subscriptionReadLock.unlock(); } } InMemoryLKVLiveMarketDataProvider(final LiveDataClient liveDataClient, final MarketDataAvailabilityFilter availabilityFilter, final UserPrincipal marketDataUser);  InMemoryLKVLiveMarketDataProvider(final LiveDataClient liveDataClient, final MarketDataAvailabilityFilter availabilityFilter, final MarketDataPermissionProvider permissionProvider,
      final UserPrincipal marketDataUser); @Override String getMarketDataUser(); @Override int getRequestedLiveDataSubscriptionCount(); @Override int getActiveValueSpecificationSubscriptionCount(); @Override Map<String, SubscriptionInfo> queryByTicker(String ticker); @Override void subscribe(final ValueSpecification valueSpecification); @Override void subscribe(final Set<ValueSpecification> valueSpecifications); @Override void unsubscribe(final ValueSpecification valueSpecification); @Override void unsubscribe(final Set<ValueSpecification> valueSpecifications); @Override MarketDataAvailabilityProvider getAvailabilityProvider(final MarketDataSpecification marketDataSpec); @Override MarketDataPermissionProvider getPermissionProvider(); @Override boolean isCompatible(final MarketDataSpecification marketDataSpec); @Override MarketDataSnapshot snapshot(final MarketDataSpecification marketDataSpec); @Override void subscriptionResultReceived(final LiveDataSubscriptionResponse subscriptionResult); @Override void subscriptionResultsReceived(final Collection<LiveDataSubscriptionResponse> subscriptionResults); @Override boolean isActive(final ValueSpecification specification); @Override void subscriptionStopped(final LiveDataSpecification fullyQualifiedSpecification); @Override void valueUpdate(final LiveDataValueUpdate valueUpdate);  }
@Test public void testResubscribe() { TestLiveDataClient liveDataClient = new TestLiveDataClient(); InMemoryLKVLiveMarketDataProvider provider = new InMemoryLKVLiveMarketDataProvider(liveDataClient, mock(MarketDataAvailabilityFilter.class), UserPrincipal.getTestUser()); ValueSpecification primitiveSpecification = createPrimitiveValueSpec("AAPL."); provider.subscribe(primitiveSpecification); assertEquals(1, provider.getRequestedLiveDataSubscriptionCount()); assertEquals(0, provider.getActiveValueSpecificationSubscriptionCount()); assertEquals(1, liveDataClient.getSubscriptionRequests().size()); LiveDataSpecification liveDataSpecification = LiveMarketDataAvailabilityProvider.getLiveDataSpecification(primitiveSpecification); LiveDataSubscriptionResponse primitiveResponse = new LiveDataSubscriptionResponse(liveDataSpecification, LiveDataSubscriptionResult.SUCCESS, null, liveDataSpecification, "primitive", null); provider.subscriptionResultReceived(primitiveResponse); assertEquals(1, provider.getRequestedLiveDataSubscriptionCount()); assertEquals(1, provider.getActiveValueSpecificationSubscriptionCount()); assertEquals(1, liveDataClient.getSubscriptionRequests().size()); assertEquals(0, liveDataClient.getCancelRequests().size()); provider.resubscribe(ImmutableSet.of(liveDataSpecification.getIdentifiers().getExternalIds().first().getScheme())); assertEquals(1, provider.getRequestedLiveDataSubscriptionCount()); assertEquals(1, provider.getActiveValueSpecificationSubscriptionCount()); assertEquals(2, liveDataClient.getSubscriptionRequests().size()); assertEquals(0, liveDataClient.getCancelRequests().size()); provider.subscriptionResultReceived(primitiveResponse); assertEquals(1, provider.getRequestedLiveDataSubscriptionCount()); assertEquals(1, provider.getActiveValueSpecificationSubscriptionCount()); assertEquals(2, liveDataClient.getSubscriptionRequests().size()); assertEquals(0, liveDataClient.getCancelRequests().size()); }