@Test void shouldBatchMessagesBelowSizeThreshold() { MantisEvent event = new MantisEvent(1, "12345"); aggregator.write(ctx, event, ctx.channel().voidPromise()); verify(ctx, times(0)).write(any(FullHttpRequest.class), any()); }
@Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (acceptOutboundMessage(msg)) { MantisEvent event = (MantisEvent) msg; int eventSize = event.size(); if (currentMessageSize + eventSize > flushIntervalBytes) { writeBatch(ctx, promise); } else { promise.setSuccess(); } currentMessage.addEvent(event); currentMessageSize += event.size(); } else { ctx.write(msg, promise); } }
MantisEventAggregator extends ChannelOutboundHandlerAdapter { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (acceptOutboundMessage(msg)) { MantisEvent event = (MantisEvent) msg; int eventSize = event.size(); if (currentMessageSize + eventSize > flushIntervalBytes) { writeBatch(ctx, promise); } else { promise.setSuccess(); } currentMessage.addEvent(event); currentMessageSize += event.size(); } else { ctx.write(msg, promise); } } }
MantisEventAggregator extends ChannelOutboundHandlerAdapter { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (acceptOutboundMessage(msg)) { MantisEvent event = (MantisEvent) msg; int eventSize = event.size(); if (currentMessageSize + eventSize > flushIntervalBytes) { writeBatch(ctx, promise); } else { promise.setSuccess(); } currentMessage.addEvent(event); currentMessageSize += event.size(); } else { ctx.write(msg, promise); } } MantisEventAggregator(Registry registry,
                          Clock clock,
                          boolean compress,
                          long flushIntervalMs,
                          int flushIntervalBytes); }
MantisEventAggregator extends ChannelOutboundHandlerAdapter { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (acceptOutboundMessage(msg)) { MantisEvent event = (MantisEvent) msg; int eventSize = event.size(); if (currentMessageSize + eventSize > flushIntervalBytes) { writeBatch(ctx, promise); } else { promise.setSuccess(); } currentMessage.addEvent(event); currentMessageSize += event.size(); } else { ctx.write(msg, promise); } } MantisEventAggregator(Registry registry,
                          Clock clock,
                          boolean compress,
                          long flushIntervalMs,
                          int flushIntervalBytes); @Override void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void handlerRemoved(ChannelHandlerContext ctx); }
MantisEventAggregator extends ChannelOutboundHandlerAdapter { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (acceptOutboundMessage(msg)) { MantisEvent event = (MantisEvent) msg; int eventSize = event.size(); if (currentMessageSize + eventSize > flushIntervalBytes) { writeBatch(ctx, promise); } else { promise.setSuccess(); } currentMessage.addEvent(event); currentMessageSize += event.size(); } else { ctx.write(msg, promise); } } MantisEventAggregator(Registry registry,
                          Clock clock,
                          boolean compress,
                          long flushIntervalMs,
                          int flushIntervalBytes); @Override void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void handlerRemoved(ChannelHandlerContext ctx);  }