@Test(timeout = 60000) public void testIncompleteFieldName() throws Exception { ServerSentEvent expected = newServerSentEvent("add", null, "data line"); List<Object> out = new ArrayList<>(); decoder.decode(ch, toHttpContent("ev"), out); assertEquals("Unexpected number of decoded messages.", 0, out.size()); decoder.decode(ch, toHttpContent("ent: add\n d"), out); assertEquals("Unexpected number of decoded messages.", 0, out.size()); doTest("ata: data line\n", expected); }
@Override protected void decode(ChannelHandlerContext ctx, HttpContent httpContent, List<Object> out) throws Exception { final ByteBuf in = httpContent.content(); if (null == sseEncodingCharset) { throw new IllegalArgumentException("Can not read SSE data as UTF-8 charset is not available."); } while (in.isReadable()) { final int readerIndexAtStart = in.readerIndex(); switch (state) { case SkipColonAndWhiteSpaces: if (skipColonAndWhiteSpaces(in)) { state = State.ReadFieldValue; } break; case SkipLineDelimitersAndSpaces: if (skipLineDelimiters(in)) { state = State.ReadFieldName; } break; case DiscardTillEOL: if(skipTillEOL(in)) { state = State.SkipLineDelimitersAndSpaces; } break; case ReadFieldName: final int indexOfColon = scanAndFindColon(in); if (-1 == indexOfColon) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(); } incompleteData.writeBytes(in); } else { int fieldNameLengthInTheCurrentBuffer = indexOfColon - readerIndexAtStart; ByteBuf fieldNameBuffer; if (null != incompleteData) { in.readBytes(incompleteData, fieldNameLengthInTheCurrentBuffer); fieldNameBuffer = incompleteData; incompleteData = null; } else { fieldNameBuffer = ctx.alloc().buffer(fieldNameLengthInTheCurrentBuffer, fieldNameLengthInTheCurrentBuffer); in.readBytes(fieldNameBuffer, fieldNameLengthInTheCurrentBuffer); } state = State.SkipColonAndWhiteSpaces; try { currentFieldType = readCurrentFieldTypeFromBuffer(fieldNameBuffer); } finally { if (null == currentFieldType) { state = State.DiscardTillEOL; } fieldNameBuffer.release(); } } break; case ReadFieldValue: final int endOfLineStartIndex = scanAndFindEndOfLine(in); if (-1 == endOfLineStartIndex) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(in.readableBytes()); } incompleteData.writeBytes(in); } else { final int bytesAvailableInThisIteration = endOfLineStartIndex - readerIndexAtStart; if (null == incompleteData) { incompleteData = ctx.alloc().buffer(bytesAvailableInThisIteration, bytesAvailableInThisIteration); } incompleteData.writeBytes(in, bytesAvailableInThisIteration); switch (currentFieldType) { case Data: if (incompleteData.isReadable()) { out.add(ServerSentEvent.withEventIdAndType(lastEventId, lastEventType, incompleteData)); } else { incompleteData.release(); } break; case Id: if (incompleteData.isReadable()) { lastEventId = incompleteData; } else { incompleteData.release(); lastEventId = null; } break; case EventType: if (incompleteData.isReadable()) { lastEventType = incompleteData; } else { incompleteData.release(); lastEventType = null; } break; } incompleteData = null; state = State.SkipLineDelimitersAndSpaces; } break; } } if (httpContent instanceof LastHttpContent) { ctx.fireChannelRead(httpContent); } }
ServerSentEventDecoder extends MessageToMessageDecoder<HttpContent> { @Override protected void decode(ChannelHandlerContext ctx, HttpContent httpContent, List<Object> out) throws Exception { final ByteBuf in = httpContent.content(); if (null == sseEncodingCharset) { throw new IllegalArgumentException("Can not read SSE data as UTF-8 charset is not available."); } while (in.isReadable()) { final int readerIndexAtStart = in.readerIndex(); switch (state) { case SkipColonAndWhiteSpaces: if (skipColonAndWhiteSpaces(in)) { state = State.ReadFieldValue; } break; case SkipLineDelimitersAndSpaces: if (skipLineDelimiters(in)) { state = State.ReadFieldName; } break; case DiscardTillEOL: if(skipTillEOL(in)) { state = State.SkipLineDelimitersAndSpaces; } break; case ReadFieldName: final int indexOfColon = scanAndFindColon(in); if (-1 == indexOfColon) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(); } incompleteData.writeBytes(in); } else { int fieldNameLengthInTheCurrentBuffer = indexOfColon - readerIndexAtStart; ByteBuf fieldNameBuffer; if (null != incompleteData) { in.readBytes(incompleteData, fieldNameLengthInTheCurrentBuffer); fieldNameBuffer = incompleteData; incompleteData = null; } else { fieldNameBuffer = ctx.alloc().buffer(fieldNameLengthInTheCurrentBuffer, fieldNameLengthInTheCurrentBuffer); in.readBytes(fieldNameBuffer, fieldNameLengthInTheCurrentBuffer); } state = State.SkipColonAndWhiteSpaces; try { currentFieldType = readCurrentFieldTypeFromBuffer(fieldNameBuffer); } finally { if (null == currentFieldType) { state = State.DiscardTillEOL; } fieldNameBuffer.release(); } } break; case ReadFieldValue: final int endOfLineStartIndex = scanAndFindEndOfLine(in); if (-1 == endOfLineStartIndex) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(in.readableBytes()); } incompleteData.writeBytes(in); } else { final int bytesAvailableInThisIteration = endOfLineStartIndex - readerIndexAtStart; if (null == incompleteData) { incompleteData = ctx.alloc().buffer(bytesAvailableInThisIteration, bytesAvailableInThisIteration); } incompleteData.writeBytes(in, bytesAvailableInThisIteration); switch (currentFieldType) { case Data: if (incompleteData.isReadable()) { out.add(ServerSentEvent.withEventIdAndType(lastEventId, lastEventType, incompleteData)); } else { incompleteData.release(); } break; case Id: if (incompleteData.isReadable()) { lastEventId = incompleteData; } else { incompleteData.release(); lastEventId = null; } break; case EventType: if (incompleteData.isReadable()) { lastEventType = incompleteData; } else { incompleteData.release(); lastEventType = null; } break; } incompleteData = null; state = State.SkipLineDelimitersAndSpaces; } break; } } if (httpContent instanceof LastHttpContent) { ctx.fireChannelRead(httpContent); } } }
ServerSentEventDecoder extends MessageToMessageDecoder<HttpContent> { @Override protected void decode(ChannelHandlerContext ctx, HttpContent httpContent, List<Object> out) throws Exception { final ByteBuf in = httpContent.content(); if (null == sseEncodingCharset) { throw new IllegalArgumentException("Can not read SSE data as UTF-8 charset is not available."); } while (in.isReadable()) { final int readerIndexAtStart = in.readerIndex(); switch (state) { case SkipColonAndWhiteSpaces: if (skipColonAndWhiteSpaces(in)) { state = State.ReadFieldValue; } break; case SkipLineDelimitersAndSpaces: if (skipLineDelimiters(in)) { state = State.ReadFieldName; } break; case DiscardTillEOL: if(skipTillEOL(in)) { state = State.SkipLineDelimitersAndSpaces; } break; case ReadFieldName: final int indexOfColon = scanAndFindColon(in); if (-1 == indexOfColon) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(); } incompleteData.writeBytes(in); } else { int fieldNameLengthInTheCurrentBuffer = indexOfColon - readerIndexAtStart; ByteBuf fieldNameBuffer; if (null != incompleteData) { in.readBytes(incompleteData, fieldNameLengthInTheCurrentBuffer); fieldNameBuffer = incompleteData; incompleteData = null; } else { fieldNameBuffer = ctx.alloc().buffer(fieldNameLengthInTheCurrentBuffer, fieldNameLengthInTheCurrentBuffer); in.readBytes(fieldNameBuffer, fieldNameLengthInTheCurrentBuffer); } state = State.SkipColonAndWhiteSpaces; try { currentFieldType = readCurrentFieldTypeFromBuffer(fieldNameBuffer); } finally { if (null == currentFieldType) { state = State.DiscardTillEOL; } fieldNameBuffer.release(); } } break; case ReadFieldValue: final int endOfLineStartIndex = scanAndFindEndOfLine(in); if (-1 == endOfLineStartIndex) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(in.readableBytes()); } incompleteData.writeBytes(in); } else { final int bytesAvailableInThisIteration = endOfLineStartIndex - readerIndexAtStart; if (null == incompleteData) { incompleteData = ctx.alloc().buffer(bytesAvailableInThisIteration, bytesAvailableInThisIteration); } incompleteData.writeBytes(in, bytesAvailableInThisIteration); switch (currentFieldType) { case Data: if (incompleteData.isReadable()) { out.add(ServerSentEvent.withEventIdAndType(lastEventId, lastEventType, incompleteData)); } else { incompleteData.release(); } break; case Id: if (incompleteData.isReadable()) { lastEventId = incompleteData; } else { incompleteData.release(); lastEventId = null; } break; case EventType: if (incompleteData.isReadable()) { lastEventType = incompleteData; } else { incompleteData.release(); lastEventType = null; } break; } incompleteData = null; state = State.SkipLineDelimitersAndSpaces; } break; } } if (httpContent instanceof LastHttpContent) { ctx.fireChannelRead(httpContent); } }  }
ServerSentEventDecoder extends MessageToMessageDecoder<HttpContent> { @Override protected void decode(ChannelHandlerContext ctx, HttpContent httpContent, List<Object> out) throws Exception { final ByteBuf in = httpContent.content(); if (null == sseEncodingCharset) { throw new IllegalArgumentException("Can not read SSE data as UTF-8 charset is not available."); } while (in.isReadable()) { final int readerIndexAtStart = in.readerIndex(); switch (state) { case SkipColonAndWhiteSpaces: if (skipColonAndWhiteSpaces(in)) { state = State.ReadFieldValue; } break; case SkipLineDelimitersAndSpaces: if (skipLineDelimiters(in)) { state = State.ReadFieldName; } break; case DiscardTillEOL: if(skipTillEOL(in)) { state = State.SkipLineDelimitersAndSpaces; } break; case ReadFieldName: final int indexOfColon = scanAndFindColon(in); if (-1 == indexOfColon) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(); } incompleteData.writeBytes(in); } else { int fieldNameLengthInTheCurrentBuffer = indexOfColon - readerIndexAtStart; ByteBuf fieldNameBuffer; if (null != incompleteData) { in.readBytes(incompleteData, fieldNameLengthInTheCurrentBuffer); fieldNameBuffer = incompleteData; incompleteData = null; } else { fieldNameBuffer = ctx.alloc().buffer(fieldNameLengthInTheCurrentBuffer, fieldNameLengthInTheCurrentBuffer); in.readBytes(fieldNameBuffer, fieldNameLengthInTheCurrentBuffer); } state = State.SkipColonAndWhiteSpaces; try { currentFieldType = readCurrentFieldTypeFromBuffer(fieldNameBuffer); } finally { if (null == currentFieldType) { state = State.DiscardTillEOL; } fieldNameBuffer.release(); } } break; case ReadFieldValue: final int endOfLineStartIndex = scanAndFindEndOfLine(in); if (-1 == endOfLineStartIndex) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(in.readableBytes()); } incompleteData.writeBytes(in); } else { final int bytesAvailableInThisIteration = endOfLineStartIndex - readerIndexAtStart; if (null == incompleteData) { incompleteData = ctx.alloc().buffer(bytesAvailableInThisIteration, bytesAvailableInThisIteration); } incompleteData.writeBytes(in, bytesAvailableInThisIteration); switch (currentFieldType) { case Data: if (incompleteData.isReadable()) { out.add(ServerSentEvent.withEventIdAndType(lastEventId, lastEventType, incompleteData)); } else { incompleteData.release(); } break; case Id: if (incompleteData.isReadable()) { lastEventId = incompleteData; } else { incompleteData.release(); lastEventId = null; } break; case EventType: if (incompleteData.isReadable()) { lastEventType = incompleteData; } else { incompleteData.release(); lastEventType = null; } break; } incompleteData = null; state = State.SkipLineDelimitersAndSpaces; } break; } } if (httpContent instanceof LastHttpContent) { ctx.fireChannelRead(httpContent); } }  @Override void channelInactive(ChannelHandlerContext ctx); }
ServerSentEventDecoder extends MessageToMessageDecoder<HttpContent> { @Override protected void decode(ChannelHandlerContext ctx, HttpContent httpContent, List<Object> out) throws Exception { final ByteBuf in = httpContent.content(); if (null == sseEncodingCharset) { throw new IllegalArgumentException("Can not read SSE data as UTF-8 charset is not available."); } while (in.isReadable()) { final int readerIndexAtStart = in.readerIndex(); switch (state) { case SkipColonAndWhiteSpaces: if (skipColonAndWhiteSpaces(in)) { state = State.ReadFieldValue; } break; case SkipLineDelimitersAndSpaces: if (skipLineDelimiters(in)) { state = State.ReadFieldName; } break; case DiscardTillEOL: if(skipTillEOL(in)) { state = State.SkipLineDelimitersAndSpaces; } break; case ReadFieldName: final int indexOfColon = scanAndFindColon(in); if (-1 == indexOfColon) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(); } incompleteData.writeBytes(in); } else { int fieldNameLengthInTheCurrentBuffer = indexOfColon - readerIndexAtStart; ByteBuf fieldNameBuffer; if (null != incompleteData) { in.readBytes(incompleteData, fieldNameLengthInTheCurrentBuffer); fieldNameBuffer = incompleteData; incompleteData = null; } else { fieldNameBuffer = ctx.alloc().buffer(fieldNameLengthInTheCurrentBuffer, fieldNameLengthInTheCurrentBuffer); in.readBytes(fieldNameBuffer, fieldNameLengthInTheCurrentBuffer); } state = State.SkipColonAndWhiteSpaces; try { currentFieldType = readCurrentFieldTypeFromBuffer(fieldNameBuffer); } finally { if (null == currentFieldType) { state = State.DiscardTillEOL; } fieldNameBuffer.release(); } } break; case ReadFieldValue: final int endOfLineStartIndex = scanAndFindEndOfLine(in); if (-1 == endOfLineStartIndex) { if (null == incompleteData) { incompleteData = ctx.alloc().buffer(in.readableBytes()); } incompleteData.writeBytes(in); } else { final int bytesAvailableInThisIteration = endOfLineStartIndex - readerIndexAtStart; if (null == incompleteData) { incompleteData = ctx.alloc().buffer(bytesAvailableInThisIteration, bytesAvailableInThisIteration); } incompleteData.writeBytes(in, bytesAvailableInThisIteration); switch (currentFieldType) { case Data: if (incompleteData.isReadable()) { out.add(ServerSentEvent.withEventIdAndType(lastEventId, lastEventType, incompleteData)); } else { incompleteData.release(); } break; case Id: if (incompleteData.isReadable()) { lastEventId = incompleteData; } else { incompleteData.release(); lastEventId = null; } break; case EventType: if (incompleteData.isReadable()) { lastEventType = incompleteData; } else { incompleteData.release(); lastEventType = null; } break; } incompleteData = null; state = State.SkipLineDelimitersAndSpaces; } break; } } if (httpContent instanceof LastHttpContent) { ctx.fireChannelRead(httpContent); } }  @Override void channelInactive(ChannelHandlerContext ctx); static final int DEFAULT_MAX_FIELD_LENGTH; }