EventSourceTopology implements EventAggregatorService, EventSourceRegistry { @Override public ListenableFuture<RpcResult<DestroyTopicOutput>> destroyTopic(final DestroyTopicInput input) { final EventSourceTopic topicToDestroy = eventSourceTopicMap.remove(input.getTopicId()); if (topicToDestroy != null) { topicToDestroy.close(); } return Util.resultRpcSuccessFor(new DestroyTopicOutputBuilder().build()); } 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 destroyTopicTest() throws Exception { topicTestHelper(); TopicId topicId = new TopicId("topic-id-007"); Map<TopicId, EventSourceTopic> localMap = eventSourceTopology.getEventSourceTopicMap(); EventSourceTopic eventSourceTopic = EventSourceTopic.create(new NotificationPattern("foo"), "pattern", eventSourceTopology); localMap.put(topicId, eventSourceTopic); DestroyTopicInput input = new DestroyTopicInputBuilder().setTopicId(topicId).build(); eventSourceTopology.destroyTopic(input); verify(listenerRegistrationMock, times(1)).close(); }