@Test(timeout = 10000L) public void testNettyTextSink() throws Exception { final int numSinks = 4; final List<String> outputStream = Arrays.asList( "Lorem ipsum dolor sit amet, consectetur adipiscing elit.", "In in leo nec erat fringilla mattis eu non massa.", "Cras quis diam suscipit, commodo enim id, pulvinar nunc."); final CountDownLatch countDownLatch = new CountDownLatch(numSinks * outputStream.size()); final CountDownLatch channelCountDown = new CountDownLatch(numSinks); final Map<Channel, List<String>> channelListMap = new ConcurrentHashMap<>(); final NettyChannelHandler channelHandler = new TestReceiverChannelHandler(channelCountDown, countDownLatch, channelListMap); try (final NettyTextMessageOutputReceiver outputReceiver = new NettyTextMessageOutputReceiver(SERVER_ADDR, SERVER_PORT, channelHandler)) { final List<Sink<String>> sinks = new LinkedList<>(); for (int i = 0; i < numSinks; i++) { final Sink<String> sink = new NettyTextSink(SERVER_ADDR, SERVER_PORT, nettySharedResource, stringIdentifierFactory); sinks.add(sink); } channelCountDown.await(); outputStream.forEach((output) -> { for (final Sink<String> sink : sinks) { sink.handle(output); } }); countDownLatch.await(); for (List<String> received : channelListMap.values()) { Assert.assertEquals(outputStream, received); } for (final Sink<String> sink : sinks) { sink.close(); } } }
@Inject public NettyTextSink( @Parameter(SocketServerIp.class) final String serverAddress, @Parameter(SocketServerPort.class) final int port, final NettySharedResource sharedResource, final StringIdentifierFactory identifierFactory) throws IOException { final Bootstrap clientBootstrap = sharedResource.getClientBootstrap(); final ChannelFuture channelFuture = clientBootstrap.connect(serverAddress, port); channelFuture.awaitUninterruptibly(); assert channelFuture.isDone(); if (!channelFuture.isSuccess()) { final StringBuilder sb = new StringBuilder("A connection failed at Sink - "); sb.append(channelFuture.cause()); throw new RuntimeException(sb.toString()); } this.channel = channelFuture.channel(); }
NettyTextSink implements Sink<String> { @Inject public NettyTextSink( @Parameter(SocketServerIp.class) final String serverAddress, @Parameter(SocketServerPort.class) final int port, final NettySharedResource sharedResource, final StringIdentifierFactory identifierFactory) throws IOException { final Bootstrap clientBootstrap = sharedResource.getClientBootstrap(); final ChannelFuture channelFuture = clientBootstrap.connect(serverAddress, port); channelFuture.awaitUninterruptibly(); assert channelFuture.isDone(); if (!channelFuture.isSuccess()) { final StringBuilder sb = new StringBuilder("A connection failed at Sink - "); sb.append(channelFuture.cause()); throw new RuntimeException(sb.toString()); } this.channel = channelFuture.channel(); } }
NettyTextSink implements Sink<String> { @Inject public NettyTextSink( @Parameter(SocketServerIp.class) final String serverAddress, @Parameter(SocketServerPort.class) final int port, final NettySharedResource sharedResource, final StringIdentifierFactory identifierFactory) throws IOException { final Bootstrap clientBootstrap = sharedResource.getClientBootstrap(); final ChannelFuture channelFuture = clientBootstrap.connect(serverAddress, port); channelFuture.awaitUninterruptibly(); assert channelFuture.isDone(); if (!channelFuture.isSuccess()) { final StringBuilder sb = new StringBuilder("A connection failed at Sink - "); sb.append(channelFuture.cause()); throw new RuntimeException(sb.toString()); } this.channel = channelFuture.channel(); } @Inject  NettyTextSink(
      @Parameter(SocketServerIp.class) final String serverAddress,
      @Parameter(SocketServerPort.class) final int port,
      final NettySharedResource sharedResource,
      final StringIdentifierFactory identifierFactory); }
NettyTextSink implements Sink<String> { @Inject public NettyTextSink( @Parameter(SocketServerIp.class) final String serverAddress, @Parameter(SocketServerPort.class) final int port, final NettySharedResource sharedResource, final StringIdentifierFactory identifierFactory) throws IOException { final Bootstrap clientBootstrap = sharedResource.getClientBootstrap(); final ChannelFuture channelFuture = clientBootstrap.connect(serverAddress, port); channelFuture.awaitUninterruptibly(); assert channelFuture.isDone(); if (!channelFuture.isSuccess()) { final StringBuilder sb = new StringBuilder("A connection failed at Sink - "); sb.append(channelFuture.cause()); throw new RuntimeException(sb.toString()); } this.channel = channelFuture.channel(); } @Inject  NettyTextSink(
      @Parameter(SocketServerIp.class) final String serverAddress,
      @Parameter(SocketServerPort.class) final int port,
      final NettySharedResource sharedResource,
      final StringIdentifierFactory identifierFactory); @Override void close(); @Override void handle(final String input); }
NettyTextSink implements Sink<String> { @Inject public NettyTextSink( @Parameter(SocketServerIp.class) final String serverAddress, @Parameter(SocketServerPort.class) final int port, final NettySharedResource sharedResource, final StringIdentifierFactory identifierFactory) throws IOException { final Bootstrap clientBootstrap = sharedResource.getClientBootstrap(); final ChannelFuture channelFuture = clientBootstrap.connect(serverAddress, port); channelFuture.awaitUninterruptibly(); assert channelFuture.isDone(); if (!channelFuture.isSuccess()) { final StringBuilder sb = new StringBuilder("A connection failed at Sink - "); sb.append(channelFuture.cause()); throw new RuntimeException(sb.toString()); } this.channel = channelFuture.channel(); } @Inject  NettyTextSink(
      @Parameter(SocketServerIp.class) final String serverAddress,
      @Parameter(SocketServerPort.class) final int port,
      final NettySharedResource sharedResource,
      final StringIdentifierFactory identifierFactory); @Override void close(); @Override void handle(final String input);  }