@Test public void write() { final EmbeddedChannel ch = newEventSourceChannel(); ch.writeOutbound(new OpenFrame()); final HttpResponse response = ch.readOutbound(); assertThat(response.getStatus(), equalTo(HttpResponseStatus.OK)); assertThat(response.headers().get(CONTENT_TYPE), equalTo(EventSourceTransport.CONTENT_TYPE_EVENT_STREAM)); SockJsTestUtil.verifyNoCacheHeaders(response); final DefaultHttpContent newLinePrelude = ch.readOutbound(); assertThat(newLinePrelude.content().toString(UTF_8), equalTo("\r\n")); final DefaultHttpContent data = ch.readOutbound(); assertThat(data.content().toString(UTF_8), equalTo("data: o\r\n\r\n")); }
@Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws Exception { if (msg instanceof Frame) { final Frame frame = (Frame) msg; if (headerSent.compareAndSet(false, true)) { ctx.write(createResponse(CONTENT_TYPE_EVENT_STREAM), promise); ctx.writeAndFlush(new DefaultHttpContent(CRLF.duplicate())); } final ByteBuf data = ctx.alloc().buffer(); data.writeBytes(FRAME_START.duplicate()); data.writeBytes(frame.content()); data.writeBytes(FRAME_END.duplicate()); final int dataSize = data.readableBytes(); ctx.writeAndFlush(new DefaultHttpContent(data)); frame.release(); if (maxBytesLimit(dataSize)) { if (logger.isDebugEnabled()) { logger.debug("max bytesSize limit reached [{}]", config.maxStreamingBytesSize()); } ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE); } } }
EventSourceTransport extends ChannelHandlerAdapter { @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws Exception { if (msg instanceof Frame) { final Frame frame = (Frame) msg; if (headerSent.compareAndSet(false, true)) { ctx.write(createResponse(CONTENT_TYPE_EVENT_STREAM), promise); ctx.writeAndFlush(new DefaultHttpContent(CRLF.duplicate())); } final ByteBuf data = ctx.alloc().buffer(); data.writeBytes(FRAME_START.duplicate()); data.writeBytes(frame.content()); data.writeBytes(FRAME_END.duplicate()); final int dataSize = data.readableBytes(); ctx.writeAndFlush(new DefaultHttpContent(data)); frame.release(); if (maxBytesLimit(dataSize)) { if (logger.isDebugEnabled()) { logger.debug("max bytesSize limit reached [{}]", config.maxStreamingBytesSize()); } ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE); } } } }
EventSourceTransport extends ChannelHandlerAdapter { @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws Exception { if (msg instanceof Frame) { final Frame frame = (Frame) msg; if (headerSent.compareAndSet(false, true)) { ctx.write(createResponse(CONTENT_TYPE_EVENT_STREAM), promise); ctx.writeAndFlush(new DefaultHttpContent(CRLF.duplicate())); } final ByteBuf data = ctx.alloc().buffer(); data.writeBytes(FRAME_START.duplicate()); data.writeBytes(frame.content()); data.writeBytes(FRAME_END.duplicate()); final int dataSize = data.readableBytes(); ctx.writeAndFlush(new DefaultHttpContent(data)); frame.release(); if (maxBytesLimit(dataSize)) { if (logger.isDebugEnabled()) { logger.debug("max bytesSize limit reached [{}]", config.maxStreamingBytesSize()); } ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE); } } } EventSourceTransport(final SockJsConfig config, final HttpRequest request); }
EventSourceTransport extends ChannelHandlerAdapter { @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws Exception { if (msg instanceof Frame) { final Frame frame = (Frame) msg; if (headerSent.compareAndSet(false, true)) { ctx.write(createResponse(CONTENT_TYPE_EVENT_STREAM), promise); ctx.writeAndFlush(new DefaultHttpContent(CRLF.duplicate())); } final ByteBuf data = ctx.alloc().buffer(); data.writeBytes(FRAME_START.duplicate()); data.writeBytes(frame.content()); data.writeBytes(FRAME_END.duplicate()); final int dataSize = data.readableBytes(); ctx.writeAndFlush(new DefaultHttpContent(data)); frame.release(); if (maxBytesLimit(dataSize)) { if (logger.isDebugEnabled()) { logger.debug("max bytesSize limit reached [{}]", config.maxStreamingBytesSize()); } ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE); } } } EventSourceTransport(final SockJsConfig config, final HttpRequest request); @Override void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise); }
EventSourceTransport extends ChannelHandlerAdapter { @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws Exception { if (msg instanceof Frame) { final Frame frame = (Frame) msg; if (headerSent.compareAndSet(false, true)) { ctx.write(createResponse(CONTENT_TYPE_EVENT_STREAM), promise); ctx.writeAndFlush(new DefaultHttpContent(CRLF.duplicate())); } final ByteBuf data = ctx.alloc().buffer(); data.writeBytes(FRAME_START.duplicate()); data.writeBytes(frame.content()); data.writeBytes(FRAME_END.duplicate()); final int dataSize = data.readableBytes(); ctx.writeAndFlush(new DefaultHttpContent(data)); frame.release(); if (maxBytesLimit(dataSize)) { if (logger.isDebugEnabled()) { logger.debug("max bytesSize limit reached [{}]", config.maxStreamingBytesSize()); } ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE); } } } EventSourceTransport(final SockJsConfig config, final HttpRequest request); @Override void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise); static final String CONTENT_TYPE_EVENT_STREAM; }