### Question:
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);  }### Answer:
@Test void shouldForwardMessageAboveSizeThreshold() { MantisEvent event = new MantisEvent(1, "12345"); aggregator.write(ctx, event, ctx.channel().voidPromise()); aggregator.write(ctx, event, ctx.channel().voidPromise()); verify(ctx, times(1)).writeAndFlush(any(FullHttpRequest.class)); }
@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()); }
@Test void shouldForwardUnacceptedMessage() { aggregator.write(ctx, new byte[1], ctx.channel().voidPromise()); verify(ctx, times(1)).write(any(), any()); }
@Test void shouldForwardOversizedMessages() { MantisEvent event = new MantisEvent(1, Arrays.toString(new byte[11])); aggregator.write(ctx, event, ctx.channel().voidPromise()); verify(ctx, times(1)).writeAndFlush(any(FullHttpRequest.class)); }