EventSourceTopology implements EventAggregatorService, EventSourceRegistry { @Override public ListenableFuture<RpcResult<CreateTopicOutput>> createTopic(final CreateTopicInput input) { LOG.debug("Received Topic creation request: NotificationPattern -> {}, NodeIdPattern -> {}", input.getNotificationPattern(), input.getNodeIdPattern()); final NotificationPattern notificationPattern = new NotificationPattern(input.getNotificationPattern()); final String nodeIdPattern = input.getNodeIdPattern().getValue(); final EventSourceTopic eventSourceTopic = EventSourceTopic.create(notificationPattern, nodeIdPattern, this); eventSourceTopicMap.put(eventSourceTopic.getTopicId(), eventSourceTopic); final CreateTopicOutput cto = new CreateTopicOutputBuilder() .setTopicId(eventSourceTopic.getTopicId()) .build(); LOG.info("Topic has been created: NotificationPattern -> {}, NodeIdPattern -> {}", input.getNotificationPattern(), input.getNodeIdPattern()); return Util.resultRpcSuccessFor(cto); } 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 createTopicTest() throws Exception { topicTestHelper(); assertNotNull("Topic has not been created correctly.", eventSourceTopology.createTopic(createTopicInputMock)); }