EventSourceTopology implements EventAggregatorService, EventSourceRegistry { @Override public void close() { aggregatorRpcReg.close(); eventSourceTopicMap.values().forEach(EventSourceTopic::close); } EventSourceTopology(final DataBroker dataBroker, final RpcProviderService providerService,
            RpcConsumerRegistry rpcService); @Override ListenableFuture<RpcResult<CreateTopicOutput>> createTopic(final CreateTopicInput input); @Override ListenableFuture<RpcResult<DestroyTopicOutput>> destroyTopic(final DestroyTopicInput input); @Override void close(); void register(final EventSource eventSource); void unRegister(final EventSource eventSource); @Override EventSourceRegistration<T> registerEventSource(final T eventSource);  }
@Test public void closeTest() throws Exception { constructorTestHelper(); topicTestHelper(); Map<TopicId, EventSourceTopic> localMap = eventSourceTopology.getEventSourceTopicMap(); TopicId topicIdMock = mock(TopicId.class); EventSourceTopic eventSourceTopic = EventSourceTopic.create(new NotificationPattern("foo"), "pattern", eventSourceTopology); localMap.put(topicIdMock, eventSourceTopic); eventSourceTopology.close(); verify(aggregatorRpcReg, times(1)).close(); verify(listenerRegistrationMock, times(1)).close(); }