Fixes buffer management issue in ServerSentEventDecoder.

The lastEventType and lastEventId buffers were reused but not retained. This would make the buffers recycled after a generated ServerSentEvent (using these buffers) is released. This leads to all sort of issues, from illegal reference count to wrong dangling pointers.

Now, retaining the buffers for lastEventId and lastEventType once (on channel remove or on receiving a new id/type) whenever they are used to create a ServerSentEvent instance. These buffers are released when the handler is removed.

Author: Nitesh Kant

rxnetty/src/main/java/io/reactivex/netty/protocol/http/sse/ServerSentEventDecoder.java

@@ -196,6 +196,12 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
                         switch (currentFieldType) {
                             case Data:
                                 if (incompleteData.isReadable()) {
+                                    if (null != lastEventId) {
+                                        lastEventId.retain();
+                                    }
+                                    if (null != lastEventType) {
+                                        lastEventType.retain();
+                                    }
                                     out.add(ServerSentEvent.withEventIdAndType(lastEventId, lastEventType,
                                                                                incompleteData));
                                 } else {
@@ -207,6 +213,9 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
                                     lastEventId = incompleteData;
                                 } else {
                                     incompleteData.release();
+                                    if (null != lastEventId) {
+                                        lastEventId.release();
+                                    }
                                     lastEventId = null;
                                 }
                                 break;
@@ -215,6 +224,9 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
                                     lastEventType = incompleteData;
                                 } else {
                                     incompleteData.release();
+                                    if (null != lastEventType) {
+                                        lastEventType.release();
+                                    }
                                     lastEventType = null;
                                 }
                                 break;