EventSourceTopology implements EventAggregatorService, EventSourceRegistry { @Override public <T extends EventSource> EventSourceRegistration<T> registerEventSource(final T eventSource) { final EventSourceRegistrationImpl<T> esr = new EventSourceRegistrationImpl<>(eventSource, this); register(eventSource); return esr; } 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 registerEventSourceTest() throws Exception { topicTestHelper(); Node nodeMock = mock(Node.class); EventSource eventSourceMock = mock(EventSource.class); NodeId nodeId = new NodeId("nodeIdValue1"); NodeKey nodeKey = new NodeKey(nodeId); doReturn(nodeKey).when(nodeMock).key(); doReturn(nodeKey).when(eventSourceMock).getSourceNodeKey(); ObjectRegistration routedRpcRegistrationMock = mock(ObjectRegistration.class); doReturn(routedRpcRegistrationMock).when(rpcProviderRegistryMock) .registerRpcImplementation(eq(EventSourceService.class), eq(eventSourceMock), any(Set.class)); assertNotNull("Return value has not been created correctly.", eventSourceTopology.registerEventSource(eventSourceMock)); }