make QuicStreamEndPoint extend from AbstractEndPoint and handle finished streams

Signed-off-by: Ludovic Orban <lorban@bitronix.be>

jetty-http3/http3-server/src/main/java/org/eclipse/jetty/http3/server/QuicSession.java

@@ -64,16 +64,19 @@ public class QuicSession
         return flushed;
     }
 
-    public boolean isInputShutdown(long streamId)
+    public boolean isFinished(long streamId)
     {
-        // TODO:
+        return quicheConnection.isStreamFinished(streamId);
-        return false;
     }
 
-    public boolean isOutputShutdown(long streamId)
+    public void sendFinished(long streamId) throws IOException
     {
-        // TODO:
+        quicheConnection.feedFinForStream(streamId);
-        return false;
+    }
+
+    public void shutdownInput(long streamId) throws IOException
+    {
+        quicheConnection.shutdownStream(streamId, false);
     }
 
     public void shutdownOutput(long streamId) throws IOException
@@ -81,6 +84,11 @@ public class QuicSession
         quicheConnection.shutdownStream(streamId, true);
     }
 
+    public void onClose(long streamId)
+    {
+        endpoints.remove(streamId);
+    }
+
     InetSocketAddress getLocalAddress()
     {
         return connection.getEndPoint().getLocalAddress();

jetty-http3/http3-server/src/main/java/org/eclipse/jetty/http3/server/QuicStreamEndPoint.java

@@ -16,50 +16,21 @@ package org.eclipse.jetty.http3.server;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
-import java.nio.channels.ReadPendingException;
-import java.nio.channels.WritePendingException;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.AbstractEndPoint;
-import org.eclipse.jetty.io.EndPoint;
-import org.eclipse.jetty.io.FillInterest;
-import org.eclipse.jetty.io.IdleTimeout;
-import org.eclipse.jetty.io.WriteFlusher;
 import org.eclipse.jetty.util.BufferUtil;
-import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.thread.Scheduler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class QuicStreamEndPoint extends IdleTimeout implements EndPoint
+public class QuicStreamEndPoint extends AbstractEndPoint
 {
     private static final Logger LOG = LoggerFactory.getLogger(QuicStreamEndPoint.class);
 
-    private final long createdTimeStamp = System.currentTimeMillis();
     private final AtomicBoolean fillable = new AtomicBoolean();
-    private final FillInterest fillInterest = new FillInterest()
-    {
-        @Override
-        protected void needsFillInterest()
-        {
-            if (fillable.getAndSet(false))
-                fillInterest.fillable();
-        }
-    };
-    private final WriteFlusher writeFlusher = new WriteFlusher(this)
-    {
-        @Override
-        protected void onIncompleteFlush()
-        {
-            // No need to do anything.
-            // See QuicSession.process().
-        }
-    };
     private final QuicSession session;
-    private Connection connection;
     private final long streamId;
-    private boolean open;
 
     public QuicStreamEndPoint(Scheduler scheduler, QuicSession session, long streamId)
     {
@@ -81,19 +52,21 @@ public class QuicStreamEndPoint extends IdleTimeout implements EndPoint
     }
 
     @Override
-    public boolean isOpen()
+    protected void doShutdownInput()
     {
-        return open;
+        try
+        {
+            session.shutdownInput(streamId);
+        }
+        catch (IOException x)
+        {
+            if (LOG.isDebugEnabled())
+                LOG.debug("error shutting down output", x);
+        }
     }
 
     @Override
-    public long getCreatedTimeStamp()
+    protected void doShutdownOutput()
-    {
-        return createdTimeStamp;
-    }
-
-    @Override
-    public void shutdownOutput()
     {
         try
         {
@@ -107,21 +80,19 @@ public class QuicStreamEndPoint extends IdleTimeout implements EndPoint
     }
 
     @Override
-    public boolean isOutputShutdown()
+    public void onClose(Throwable failure)
     {
-        return session.isOutputShutdown(streamId);
+        super.onClose(failure);
-    }
+        try
-
+        {
-    @Override
+            session.sendFinished(streamId);
-    public boolean isInputShutdown()
+        }
-    {
+        catch (IOException e)
-        return session.isInputShutdown(streamId);
+        {
-    }
+            if (LOG.isDebugEnabled())
-
+                LOG.debug("Error sending FIN on stream {}", streamId, e);
-    @Override
+        }
-    public void close(Throwable cause)
+        session.onClose(streamId);
-    {
-        onClose(cause);
     }
 
     @Override
@@ -130,6 +101,8 @@ public class QuicStreamEndPoint extends IdleTimeout implements EndPoint
         int pos = BufferUtil.flipToFill(buffer);
         int drained = session.fill(streamId, buffer);
         BufferUtil.flipToFlush(buffer, pos);
+        if (session.isFinished(streamId))
+            shutdownInput();
         return drained;
     }
 
@@ -155,7 +128,7 @@ public class QuicStreamEndPoint extends IdleTimeout implements EndPoint
 
     public void onWritable()
     {
-        writeFlusher.completeWrite();
+        getWriteFlusher().completeWrite();
     }
 
     public Runnable onReadable()
@@ -163,68 +136,22 @@ public class QuicStreamEndPoint extends IdleTimeout implements EndPoint
         return () ->
         {
             //TODO: this is racy
-            if (!fillInterest.fillable())
+            if (!getFillInterest().fillable())
                 fillable.set(true);
         };
     }
 
     @Override
-    public void fillInterested(Callback callback) throws ReadPendingException
+    protected void onIncompleteFlush()
     {
-        fillInterest.register(callback);
+        // No need to do anything.
+        // See QuicSession.process().
     }
 
     @Override
-    public boolean tryFillInterested(Callback callback)
+    protected void needsFillInterest() throws IOException
     {
-        return fillInterest.tryRegister(callback);
+        if (fillable.getAndSet(false))
-    }
+            getFillInterest().fillable();
-
-    @Override
-    public boolean isFillInterested()
-    {
-        return fillInterest.isInterested();
-    }
-
-    @Override
-    public void write(Callback callback, ByteBuffer... buffers) throws WritePendingException
-    {
-        writeFlusher.write(callback, buffers);
-    }
-
-    @Override
-    public Connection getConnection()
-    {
-        return connection;
-    }
-
-    @Override
-    public void setConnection(Connection connection)
-    {
-        this.connection = connection;
-    }
-
-    @Override
-    public void onOpen()
-    {
-        open = true;
-    }
-
-    @Override
-    public void onClose(Throwable cause)
-    {
-        open = false;
-    }
-
-    @Override
-    public void upgrade(Connection newConnection)
-    {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    protected void onIdleExpired(TimeoutException timeout)
-    {
-
     }
 }