From 7df0dfa7c2c8210514f83688d84a9cae2ecc3cf2 Mon Sep 17 00:00:00 2001
From: Lachlan Roberts <lachlan@webtide.com>
Date: Wed, 7 Oct 2020 15:41:51 +1100
Subject: [PATCH] warn if did not read until EOF once per MessageInputStream

---
 .../common/message/MessageInputStream.java    | 22 +++++++++----------
 1 file changed, 10 insertions(+), 12 deletions(-)

diff --git a/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageInputStream.java b/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageInputStream.java
index 985fbe5808f..95c1e91aa37 100644
--- a/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageInputStream.java
+++ b/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageInputStream.java
@@ -87,7 +87,6 @@ public class MessageInputStream extends InputStream implements MessageAppender
                 switch (state)
                 {
                     case CLOSED:
-                        LOG.warn("Received content after InputStream closed");
                         return;
 
                     case RESUMED:
@@ -119,11 +118,16 @@ public class MessageInputStream extends InputStream implements MessageAppender
             if (state == State.CLOSED)
                 return;
 
-            if (!buffers.isEmpty() || (activeBuffer != null && activeBuffer.hasRemaining()))
-                LOG.warn("InputStream closed without fully consuming content");
+            boolean remainingContent = (state != State.COMPLETE) ||
+                (!buffers.isEmpty() && buffers.peek() != EOF) ||
+                (activeBuffer != null && activeBuffer.hasRemaining());
+
+            if (remainingContent)
+                LOG.warn("MessageInputStream closed without fully consuming content");
 
             state = State.CLOSED;
             buffers.clear();
+            buffers.add(EOF);
         }
     }
 
@@ -155,19 +159,13 @@ public class MessageInputStream extends InputStream implements MessageAppender
 
     public void handlerComplete()
     {
+        // Close the InputStream.
+        close();
+
         // May need to resume to resume and read to the next message.
         SuspendToken resume;
         synchronized (this)
         {
-            if (state != State.CLOSED)
-            {
-                if (!buffers.isEmpty() || (activeBuffer != null && activeBuffer.hasRemaining()))
-                    LOG.warn("InputStream closed without fully consuming content");
-
-                state = State.CLOSED;
-                buffers.clear();
-            }
-
             resume = suspendToken;
             suspendToken = null;
         }