@Test void shouldNotWriteOverActiveAndUnwritableChannel() { when(channel.isActive()).thenReturn(true); when(channel.isWritable()).thenReturn(false); eventChannel.send(mantisWorker, new Event()); verify(channel, times(0)).writeAndFlush(any(MantisEvent.class)); }
@Override public CompletableFuture<Void> send(MantisWorker worker, Event event) { InetSocketAddress address = worker.toInetSocketAddress(); Channel channel = channelManager.findOrCreate(address); CompletableFuture<Void> future = new CompletableFuture<>(); if (channel.isActive()) { if (channel.isWritable()) { LOG.debug("channel is writable: {} bytes remaining", channel.bytesBeforeUnwritable()); final long nettyStart = registry.clock().wallTime(); MantisEvent mantisEvent = new MantisEvent(1, event.toJsonString()); channel.writeAndFlush(mantisEvent).addListener(f -> { if (f.isSuccess()) { writeSuccess.increment(); future.complete(null); } else { LOG.debug("failed to send event over netty channel", f.cause()); writeFailure.increment(); future.completeExceptionally(new RetryableException(f.cause().getMessage())); } }); final long nettyEnd = registry.clock().wallTime(); nettyWriteTime.record(nettyEnd - nettyStart, TimeUnit.MILLISECONDS); } else { LOG.debug("channel not writable: {} bytes before writable", channel.bytesBeforeWritable()); nettyChannelDropped.increment(); future.completeExceptionally( new RetryableException("channel not writable: " + channel.bytesBeforeWritable() + " bytes before writable")); } } else { future.completeExceptionally(new NonRetryableException("channel not active")); } return future; }
HttpEventChannel implements EventChannel { @Override public CompletableFuture<Void> send(MantisWorker worker, Event event) { InetSocketAddress address = worker.toInetSocketAddress(); Channel channel = channelManager.findOrCreate(address); CompletableFuture<Void> future = new CompletableFuture<>(); if (channel.isActive()) { if (channel.isWritable()) { LOG.debug("channel is writable: {} bytes remaining", channel.bytesBeforeUnwritable()); final long nettyStart = registry.clock().wallTime(); MantisEvent mantisEvent = new MantisEvent(1, event.toJsonString()); channel.writeAndFlush(mantisEvent).addListener(f -> { if (f.isSuccess()) { writeSuccess.increment(); future.complete(null); } else { LOG.debug("failed to send event over netty channel", f.cause()); writeFailure.increment(); future.completeExceptionally(new RetryableException(f.cause().getMessage())); } }); final long nettyEnd = registry.clock().wallTime(); nettyWriteTime.record(nettyEnd - nettyStart, TimeUnit.MILLISECONDS); } else { LOG.debug("channel not writable: {} bytes before writable", channel.bytesBeforeWritable()); nettyChannelDropped.increment(); future.completeExceptionally( new RetryableException("channel not writable: " + channel.bytesBeforeWritable() + " bytes before writable")); } } else { future.completeExceptionally(new NonRetryableException("channel not active")); } return future; } }
HttpEventChannel implements EventChannel { @Override public CompletableFuture<Void> send(MantisWorker worker, Event event) { InetSocketAddress address = worker.toInetSocketAddress(); Channel channel = channelManager.findOrCreate(address); CompletableFuture<Void> future = new CompletableFuture<>(); if (channel.isActive()) { if (channel.isWritable()) { LOG.debug("channel is writable: {} bytes remaining", channel.bytesBeforeUnwritable()); final long nettyStart = registry.clock().wallTime(); MantisEvent mantisEvent = new MantisEvent(1, event.toJsonString()); channel.writeAndFlush(mantisEvent).addListener(f -> { if (f.isSuccess()) { writeSuccess.increment(); future.complete(null); } else { LOG.debug("failed to send event over netty channel", f.cause()); writeFailure.increment(); future.completeExceptionally(new RetryableException(f.cause().getMessage())); } }); final long nettyEnd = registry.clock().wallTime(); nettyWriteTime.record(nettyEnd - nettyStart, TimeUnit.MILLISECONDS); } else { LOG.debug("channel not writable: {} bytes before writable", channel.bytesBeforeWritable()); nettyChannelDropped.increment(); future.completeExceptionally( new RetryableException("channel not writable: " + channel.bytesBeforeWritable() + " bytes before writable")); } } else { future.completeExceptionally(new NonRetryableException("channel not active")); } return future; } HttpEventChannel(
            Registry registry,
            HttpEventChannelManager channelManager); }
HttpEventChannel implements EventChannel { @Override public CompletableFuture<Void> send(MantisWorker worker, Event event) { InetSocketAddress address = worker.toInetSocketAddress(); Channel channel = channelManager.findOrCreate(address); CompletableFuture<Void> future = new CompletableFuture<>(); if (channel.isActive()) { if (channel.isWritable()) { LOG.debug("channel is writable: {} bytes remaining", channel.bytesBeforeUnwritable()); final long nettyStart = registry.clock().wallTime(); MantisEvent mantisEvent = new MantisEvent(1, event.toJsonString()); channel.writeAndFlush(mantisEvent).addListener(f -> { if (f.isSuccess()) { writeSuccess.increment(); future.complete(null); } else { LOG.debug("failed to send event over netty channel", f.cause()); writeFailure.increment(); future.completeExceptionally(new RetryableException(f.cause().getMessage())); } }); final long nettyEnd = registry.clock().wallTime(); nettyWriteTime.record(nettyEnd - nettyStart, TimeUnit.MILLISECONDS); } else { LOG.debug("channel not writable: {} bytes before writable", channel.bytesBeforeWritable()); nettyChannelDropped.increment(); future.completeExceptionally( new RetryableException("channel not writable: " + channel.bytesBeforeWritable() + " bytes before writable")); } } else { future.completeExceptionally(new NonRetryableException("channel not active")); } return future; } HttpEventChannel(
            Registry registry,
            HttpEventChannelManager channelManager); @Override CompletableFuture<Void> send(MantisWorker worker, Event event); @Override double bufferSize(MantisWorker worker); @Override void close(MantisWorker worker); }
HttpEventChannel implements EventChannel { @Override public CompletableFuture<Void> send(MantisWorker worker, Event event) { InetSocketAddress address = worker.toInetSocketAddress(); Channel channel = channelManager.findOrCreate(address); CompletableFuture<Void> future = new CompletableFuture<>(); if (channel.isActive()) { if (channel.isWritable()) { LOG.debug("channel is writable: {} bytes remaining", channel.bytesBeforeUnwritable()); final long nettyStart = registry.clock().wallTime(); MantisEvent mantisEvent = new MantisEvent(1, event.toJsonString()); channel.writeAndFlush(mantisEvent).addListener(f -> { if (f.isSuccess()) { writeSuccess.increment(); future.complete(null); } else { LOG.debug("failed to send event over netty channel", f.cause()); writeFailure.increment(); future.completeExceptionally(new RetryableException(f.cause().getMessage())); } }); final long nettyEnd = registry.clock().wallTime(); nettyWriteTime.record(nettyEnd - nettyStart, TimeUnit.MILLISECONDS); } else { LOG.debug("channel not writable: {} bytes before writable", channel.bytesBeforeWritable()); nettyChannelDropped.increment(); future.completeExceptionally( new RetryableException("channel not writable: " + channel.bytesBeforeWritable() + " bytes before writable")); } } else { future.completeExceptionally(new NonRetryableException("channel not active")); } return future; } HttpEventChannel(
            Registry registry,
            HttpEventChannelManager channelManager); @Override CompletableFuture<Void> send(MantisWorker worker, Event event); @Override double bufferSize(MantisWorker worker); @Override void close(MantisWorker worker); static final String CHANNEL_TYPE; }