[232648] Provide an HTTPConnector

https://bugs.eclipse.org/bugs/show_bug.cgi?id=232648

1 files changed, 42 insertions, 7 deletions

diff --git a/plugins/org.eclipse.net4j.http/src/org/eclipse/net4j/internal/http/HTTPConnector.java b/plugins/org.eclipse.net4j.http/src/org/eclipse/net4j/internal/http/HTTPConnector.java
index 2924a0bdfd..1307536904 100644
--- a/plugins/org.eclipse.net4j.http/src/org/eclipse/net4j/internal/http/HTTPConnector.java
+++ b/plugins/org.eclipse.net4j.http/src/org/eclipse/net4j/internal/http/HTTPConnector.java
@@ -19,7 +19,6 @@ import org.eclipse.net4j.util.security.INegotiationContext;
 
 import org.eclipse.internal.net4j.buffer.Buffer;
 import org.eclipse.internal.net4j.channel.Channel;
-import org.eclipse.internal.net4j.channel.InternalChannel;
 import org.eclipse.internal.net4j.connector.Connector;
 
 import java.util.Queue;
@@ -34,7 +33,7 @@ public abstract class HTTPConnector extends Connector implements IHTTPConnector
 
   private String connectorID;
 
-  private Queue<IBuffer> outputQueue = new ConcurrentLinkedQueue<IBuffer>();
+  private Queue<QueuedBuffer> outputQueue = new ConcurrentLinkedQueue<QueuedBuffer>();
 
   public HTTPConnector()
   {
@@ -50,21 +49,31 @@ public abstract class HTTPConnector extends Connector implements IHTTPConnector
     this.connectorID = connectorID;
   }
 
-  public Queue<IBuffer> getOutputQueue()
+  public Queue<QueuedBuffer> getOutputQueue()
   {
     return outputQueue;
   }
 
   public void multiplexChannel(IChannel channel)
   {
-    Queue<IBuffer> channelQueue = ((InternalChannel)channel).getSendQueue();
-    IBuffer buffer = channelQueue.poll();
+    IBuffer buffer;
+    long outputBufferCount;
+
+    HTTPChannel httpChannel = (HTTPChannel)channel;
+    synchronized (httpChannel)
+    {
+      Queue<IBuffer> channelQueue = httpChannel.getSendQueue();
+      buffer = channelQueue.poll();
+      outputBufferCount = httpChannel.getOutputBufferCount();
+      httpChannel.increaseOutputBufferCount();
+    }
+
     if (TRACER.isEnabled())
     {
-      TRACER.trace("Multiplexing " + ((Buffer)buffer).formatContent(true));
+      TRACER.format("Multiplexing {0} (count={1})", ((Buffer)buffer).formatContent(true), outputBufferCount);
     }
 
-    outputQueue.add(buffer);
+    outputQueue.add(new QueuedBuffer(buffer, outputBufferCount));
   }
 
   @Override
@@ -78,4 +87,30 @@ public abstract class HTTPConnector extends Connector implements IHTTPConnector
   {
     return new HTTPChannel();
   }
+
+  /**
+   * @author Eike Stepper
+   */
+  public static final class QueuedBuffer
+  {
+    private IBuffer buffer;
+
+    private long channelCount;
+
+    public QueuedBuffer(IBuffer buffer, long channelCount)
+    {
+      this.buffer = buffer;
+      this.channelCount = channelCount;
+    }
+
+    public IBuffer getBuffer()
+    {
+      return buffer;
+    }
+
+    public long getChannelCount()
+    {
+      return channelCount;
+    }
+  }
 }