DefaultLiveDataManager implements LiveDataListener, LiveDataManager { @Override public void subscribe(final LDListener client, final Set<ExternalIdBundle> tickers) { _commandQueue.submit(new Runnable() { @Override public void run() { doSubscribe(client, tickers); } }); } DefaultLiveDataManager(LiveDataClient marketDataConnection);  DefaultLiveDataManager(LiveDataClient marketDataConnection,
                               int unsubscriptionDelay); @Override ImmutableLiveDataResults snapshot(final LDListener listener); @Override void subscribe(final LDListener client,
                        final Set<ExternalIdBundle> tickers); @Override void unsubscribe(final LDListener client,
                        final Set<ExternalIdBundle> tickers); @Override void subscriptionResultReceived(LiveDataSubscriptionResponse subscriptionResult); @Override void subscriptionResultsReceived(final Collection<LiveDataSubscriptionResponse> subscriptionResponses); @Override void waitForAllData(final LDListener listener); @Override void subscriptionStopped(LiveDataSpecification fullyQualifiedSpecification); @Override void valueUpdate(final LiveDataValueUpdate valueUpdate); @Override void unregister(final LDListener listener); static final int DEFAULT_UNSUBSCRIPTION_DELAY_SECONDS; }
@Test @SuppressWarnings("unchecked") public void testSubscribeIsOnlySentOnce() throws InterruptedException { LDListener client1 = mock(LDListener.class); LDListener client2 = mock(LDListener.class); LDListener client3 = mock(LDListener.class); _manager.subscribe(client1, createIdBundles("T1", "T2")); _manager.subscribe(client2, createIdBundles("T1", "T3")); _manager.subscribe(client3, createIdBundles("T2", "T3")); Thread.sleep(500); verify(_mockLiveDataClient, times(2)) .subscribe(any(UserPrincipal.class), any(Collection.class), any(LiveDataListener.class)); verifyNoMoreInteractions(_mockLiveDataClient); }