[375939] Payload size of 21331 cause Buffer error when CDO UI is used

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

3 files changed, 1209 insertions, 1125 deletions

diff --git a/plugins/org.eclipse.net4j.jvm/src/org/eclipse/net4j/internal/jvm/JVMConnector.java b/plugins/org.eclipse.net4j.jvm/src/org/eclipse/net4j/internal/jvm/JVMConnector.java
index 5e5f39e4d9..837f33869a 100644
--- a/plugins/org.eclipse.net4j.jvm/src/org/eclipse/net4j/internal/jvm/JVMConnector.java
+++ b/plugins/org.eclipse.net4j.jvm/src/org/eclipse/net4j/internal/jvm/JVMConnector.java
@@ -1,159 +1,169 @@
-/*

- * Copyright (c) 2004 - 2012 Eike Stepper (Berlin, Germany) and others.

- * All rights reserved. This program and the accompanying materials

- * are made available under the terms of the Eclipse Public License v1.0

- * which accompanies this distribution, and is available at

- * http://www.eclipse.org/legal/epl-v10.html

- *

- * Contributors:

- *    Eike Stepper - initial API and implementation

- */

-package org.eclipse.net4j.internal.jvm;

-

-import org.eclipse.net4j.buffer.IBuffer;

-import org.eclipse.net4j.channel.ChannelException;

-import org.eclipse.net4j.internal.jvm.bundle.OM;

-import org.eclipse.net4j.internal.jvm.messages.Messages;

-import org.eclipse.net4j.jvm.IJVMConnector;

-import org.eclipse.net4j.protocol.IProtocol;

-import org.eclipse.net4j.util.om.trace.ContextTracer;

-import org.eclipse.net4j.util.security.INegotiationContext;

-

-import org.eclipse.spi.net4j.Connector;

-import org.eclipse.spi.net4j.InternalChannel;

-

-import java.util.Queue;

-

-/**

- * TODO Remove peer channels

- * 

- * @author Eike Stepper

- */

-public abstract class JVMConnector extends Connector implements IJVMConnector

-{

-  private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG, JVMConnector.class);

-

-  private JVMConnector peer;

-

-  private String name;

-

-  public JVMConnector()

-  {

-  }

-

-  public String getName()

-  {

-    return name;

-  }

-

-  public void setName(String name)

-  {

-    this.name = name;

-  }

-

-  public JVMConnector getPeer()

-  {

-    return peer;

-  }

-

-  public void setPeer(JVMConnector peer)

-  {

-    this.peer = peer;

-  }

-

-  @Override

-  public String getURL()

-  {

-    return "jvm://" + name; //$NON-NLS-1$

-  }

-

-  public void multiplexChannel(InternalChannel localChannel)

-  {

-    short channelID = localChannel.getID();

-    InternalChannel peerChannel = peer.getChannel(channelID);

-    if (peerChannel == null)

-    {

-      throw new IllegalStateException("peerChannel == null"); //$NON-NLS-1$

-    }

-

-    Queue<IBuffer> localQueue = localChannel.getSendQueue();

-    IBuffer buffer = localQueue.poll();

-    if (TRACER.isEnabled())

-    {

-      TRACER.trace("Multiplexing " + buffer.formatContent(true)); //$NON-NLS-1$

-    }

-

-    buffer.flip();

-    peerChannel.handleBufferFromMultiplexer(buffer);

-  }

-

-  @Override

-  protected INegotiationContext createNegotiationContext()

-  {

-    throw new UnsupportedOperationException();

-  }

-

-  @Override

-  protected InternalChannel createChannel()

-  {

-    return new JVMChannel();

-  }

-

-  @Override

-  protected void registerChannelWithPeer(short channelID, long timeoutIgnored, IProtocol<?> protocol)

-      throws ChannelException

-  {

-    try

-    {

-      String protocolID = protocol == null ? null : protocol.getType();

-      JVMChannel peerChannel = (JVMChannel)peer.inverseOpenChannel(channelID, protocolID);

-      if (peerChannel == null)

-      {

-        throw new ChannelException(Messages.getString("JVMConnector.2")); //$NON-NLS-1$

-      }

-

-      JVMChannel c = (JVMChannel)getChannel(channelID);

-      c.setPeer(peerChannel);

-      peerChannel.setPeer(c);

-    }

-    catch (ChannelException ex)

-    {

-      throw ex;

-    }

-    catch (Exception ex)

-    {

-      throw new ChannelException(ex);

-    }

-  }

-

-  @Override

-  protected void deregisterChannelFromPeer(InternalChannel channel) throws ChannelException

-  {

-    try

-    {

-      getPeer().inverseCloseChannel(channel.getID());

-    }

-    catch (ChannelException ex)

-    {

-      throw ex;

-    }

-    catch (Exception ex)

-    {

-      throw new ChannelException(ex);

-    }

-  }

-

-  @Override

-  protected void doBeforeActivate() throws Exception

-  {

-    super.doBeforeActivate();

-    checkState(name, "name"); //$NON-NLS-1$

-  }

-

-  @Override

-  protected void doActivate() throws Exception

-  {

-    super.doActivate();

-    leaveConnecting();

-  }

-}

+/*
+ * Copyright (c) 2004 - 2012 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ */
+package org.eclipse.net4j.internal.jvm;
+
+import org.eclipse.net4j.buffer.IBuffer;
+import org.eclipse.net4j.channel.ChannelException;
+import org.eclipse.net4j.internal.jvm.bundle.OM;
+import org.eclipse.net4j.internal.jvm.messages.Messages;
+import org.eclipse.net4j.jvm.IJVMConnector;
+import org.eclipse.net4j.protocol.IProtocol;
+import org.eclipse.net4j.util.om.trace.ContextTracer;
+import org.eclipse.net4j.util.security.INegotiationContext;
+
+import org.eclipse.spi.net4j.Connector;
+import org.eclipse.spi.net4j.InternalChannel;
+
+import java.nio.ByteBuffer;
+import java.util.Queue;
+
+/**
+ * TODO Remove peer channels
+ *
+ * @author Eike Stepper
+ */
+public abstract class JVMConnector extends Connector implements IJVMConnector
+{
+  private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG, JVMConnector.class);
+
+  private JVMConnector peer;
+
+  private String name;
+
+  public JVMConnector()
+  {
+  }
+
+  public String getName()
+  {
+    return name;
+  }
+
+  public void setName(String name)
+  {
+    this.name = name;
+  }
+
+  public JVMConnector getPeer()
+  {
+    return peer;
+  }
+
+  public void setPeer(JVMConnector peer)
+  {
+    this.peer = peer;
+  }
+
+  @Override
+  public String getURL()
+  {
+    return "jvm://" + name; //$NON-NLS-1$
+  }
+
+  public void multiplexChannel(InternalChannel localChannel)
+  {
+    short channelID = localChannel.getID();
+    InternalChannel peerChannel = peer.getChannel(channelID);
+    if (peerChannel == null)
+    {
+      throw new IllegalStateException("peerChannel == null"); //$NON-NLS-1$
+    }
+
+    Queue<IBuffer> localQueue = localChannel.getSendQueue();
+    IBuffer buffer = localQueue.poll();
+
+    ByteBuffer byteBuffer = buffer.getByteBuffer();
+    if (byteBuffer.position() == IBuffer.HEADER_SIZE)
+    {
+      // Just release this empty buffer has been written
+      buffer.release();
+      return;
+    }
+
+    if (TRACER.isEnabled())
+    {
+      TRACER.trace("Multiplexing " + buffer.formatContent(true)); //$NON-NLS-1$
+    }
+
+    buffer.flip();
+    peerChannel.handleBufferFromMultiplexer(buffer);
+  }
+
+  @Override
+  protected INegotiationContext createNegotiationContext()
+  {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  protected InternalChannel createChannel()
+  {
+    return new JVMChannel();
+  }
+
+  @Override
+  protected void registerChannelWithPeer(short channelID, long timeoutIgnored, IProtocol<?> protocol)
+      throws ChannelException
+  {
+    try
+    {
+      String protocolID = protocol == null ? null : protocol.getType();
+      JVMChannel peerChannel = (JVMChannel)peer.inverseOpenChannel(channelID, protocolID);
+      if (peerChannel == null)
+      {
+        throw new ChannelException(Messages.getString("JVMConnector.2")); //$NON-NLS-1$
+      }
+
+      JVMChannel c = (JVMChannel)getChannel(channelID);
+      c.setPeer(peerChannel);
+      peerChannel.setPeer(c);
+    }
+    catch (ChannelException ex)
+    {
+      throw ex;
+    }
+    catch (Exception ex)
+    {
+      throw new ChannelException(ex);
+    }
+  }
+
+  @Override
+  protected void deregisterChannelFromPeer(InternalChannel channel) throws ChannelException
+  {
+    try
+    {
+      getPeer().inverseCloseChannel(channel.getID());
+    }
+    catch (ChannelException ex)
+    {
+      throw ex;
+    }
+    catch (Exception ex)
+    {
+      throw new ChannelException(ex);
+    }
+  }
+
+  @Override
+  protected void doBeforeActivate() throws Exception
+  {
+    super.doBeforeActivate();
+    checkState(name, "name"); //$NON-NLS-1$
+  }
+
+  @Override
+  protected void doActivate() throws Exception
+  {
+    super.doActivate();
+    leaveConnecting();
+  }
+}
diff --git a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/TransportTest.java b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/TransportTest.java
index bfc34b499b..26e3b2bb66 100644
--- a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/TransportTest.java
+++ b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/TransportTest.java
@@ -1,534 +1,602 @@
-/*

- * Copyright (c) 2004 - 2012 Eike Stepper (Berlin, Germany) and others.

- * All rights reserved. This program and the accompanying materials

- * are made available under the terms of the Eclipse Public License v1.0

- * which accompanies this distribution, and is available at

- * http://www.eclipse.org/legal/epl-v10.html

- *

- * Contributors:

- *    Eike Stepper - initial API and implementation

- *    Teerawat Chaiyakijpichet (No Magic Asia Ltd.) - SSL

- */

-package org.eclipse.net4j.tests;

-

-import org.eclipse.net4j.Net4jUtil;

-import org.eclipse.net4j.buffer.IBuffer;

-import org.eclipse.net4j.buffer.IBufferProvider;

-import org.eclipse.net4j.channel.ChannelInputStream;

-import org.eclipse.net4j.channel.ChannelOutputStream;

-import org.eclipse.net4j.channel.IChannel;

-import org.eclipse.net4j.connector.IConnector;

-import org.eclipse.net4j.tests.data.HugeData;

-import org.eclipse.net4j.util.container.IContainerDelta;

-import org.eclipse.net4j.util.container.IContainerEvent;

-import org.eclipse.net4j.util.event.IEvent;

-import org.eclipse.net4j.util.event.IListener;

-import org.eclipse.net4j.util.factory.IFactory;

-import org.eclipse.net4j.util.factory.ProductCreationException;

-import org.eclipse.net4j.util.io.IOUtil;

-

-import org.eclipse.spi.net4j.ClientProtocolFactory;

-import org.eclipse.spi.net4j.Connector;

-import org.eclipse.spi.net4j.Protocol;

-import org.eclipse.spi.net4j.ServerProtocolFactory;

-

-import java.io.BufferedReader;

-import java.io.DataInputStream;

-import java.io.DataOutputStream;

-import java.io.IOException;

-import java.io.InputStreamReader;

-import java.io.PrintStream;

-import java.nio.ByteBuffer;

-import java.util.StringTokenizer;

-import java.util.concurrent.CountDownLatch;

-import java.util.concurrent.TimeUnit;

-

-/**

- * @author Eike Stepper

- */

-public abstract class TransportTest extends AbstractProtocolTest

-{

-  public TransportTest()

-  {

-  }

-

-  @Override

-  protected abstract boolean useJVMTransport();

-

-  protected IBuffer provideBuffer()

-  {

-    return provideBuffer(null);

-  }

-

-  protected IBuffer provideBuffer(IConnector iConnector)

-  {

-    IBuffer buffer = null;

-    if (!useJVMTransport() && useSSLTransport())

-    {

-      // cannot use buffer provider from net4j need to use SSL Buffer inside the SSLConnector.

-      buffer = ((Connector)iConnector).provideBuffer();

-    }

-    else

-    {

-      IBufferProvider bufferProvider = Net4jUtil.getBufferProvider(container);

-      buffer = bufferProvider.provideBuffer();

-    }

-

-    return buffer;

-  }

-

-  private void registerClientFactory(IFactory factory)

-  {

-    if (!useJVMTransport() && useSSLTransport())

-    {

-      // need separate container between client and server for SSL.

-      separateContainer.registerFactory(factory);

-    }

-    else

-    {

-      container.registerFactory(factory);

-    }

-  }

-

-  protected IBufferProvider provideBufferProvider(IConnector iConnector)

-  {

-    IBufferProvider bufferProvider = null;

-    if (!useJVMTransport() && useSSLTransport())

-    {

-      // cannot use buffer provider from net4j need to use SSL Buffer inside the SSLConnector.

-      bufferProvider = ((Connector)iConnector).getConfig().getBufferProvider();

-    }

-    else

-    {

-      bufferProvider = Net4jUtil.getBufferProvider(container);

-    }

-

-    return bufferProvider;

-  }

-

-  public void testConnect() throws Exception

-  {

-    startTransport();

-  }

-

-  public void testSendBuffer() throws Exception

-  {

-    startTransport();

-    IConnector iConnecter = getConnector();

-    IChannel channel = iConnecter.openChannel();

-    for (int i = 0; i < 3; i++)

-    {

-      IBuffer buffer = provideBuffer(iConnecter);

-

-      ByteBuffer byteBuffer = buffer.startPutting(channel.getID());

-      byteBuffer.putInt(1970);

-      channel.sendBuffer(buffer);

-    }

-  }

-

-  public void testHandleBuffer() throws Exception

-  {

-    final int COUNT = 3;

-    final CountDownLatch counter = new CountDownLatch(COUNT);

-    container.registerFactory(new TestProtocol.ServerFactory(counter));

-    // need to handle about separating container between client and server for SSL.

-    registerClientFactory(new TestProtocol.ClientFactory());

-    startTransport();

-    IConnector iConnecter = getConnector();

-    IChannel channel = iConnecter.openChannel(TestProtocol.ClientFactory.TYPE, null);

-    for (int i = 0; i < COUNT; i++)

-    {

-      IBuffer buffer = provideBuffer(iConnecter);

-      ByteBuffer byteBuffer = buffer.startPutting(channel.getID());

-      byteBuffer.putInt(1970);

-      channel.sendBuffer(buffer);

-      sleep(50);

-    }

-

-    assertEquals(true, counter.await(2, TimeUnit.SECONDS));

-  }

-

-  public void testStreaming() throws Exception

-  {

-    final int COUNT = 1;

-    final CountDownLatch counter = new CountDownLatch(COUNT);

-    final ChannelInputStream[] inputStream = new ChannelInputStream[1];

-

-    getAcceptor().addListener(new IListener()

-    {

-      public void notifyEvent(IEvent event)

-      {

-        if (event instanceof IContainerEvent<?>)

-        {

-          @SuppressWarnings("unchecked")

-          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;

-          e.getDeltaElement().addListener(new IListener()

-          {

-            public void notifyEvent(IEvent event)

-            {

-              if (event instanceof IContainerEvent<?>)

-              {

-                @SuppressWarnings("unchecked")

-                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;

-                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)

-                {

-                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);

-                  counter.countDown();

-                }

-              }

-            }

-          });

-        }

-      }

-    });

-

-    IChannel channel = getConnector().openChannel();

-    assertEquals(true, counter.await(2, TimeUnit.SECONDS));

-    assertNotNull(inputStream[0]);

-

-    ChannelOutputStream outputStream = new ChannelOutputStream(channel);

-    outputStream.write(HugeData.getBytes());

-    outputStream.flushWithEOS();

-    outputStream.close();

-

-    try

-    {

-      InputStreamReader isr = new InputStreamReader(inputStream[0]);

-      BufferedReader reader = new BufferedReader(isr);

-      String line;

-      while ((line = reader.readLine()) != null)

-      {

-        msg(line);

-      }

-

-      isr.close();

-    }

-    catch (RuntimeException ex)

-    {

-      IOUtil.print(ex);

-    }

-  }

-

-  /**

-   * TODO Fails occasionally ;-( Caused by: java.lang.IllegalStateException: selectionKey == null

-   */

-  public void testTextStreaming() throws Exception

-  {

-    final int COUNT = 1;

-    final CountDownLatch counter = new CountDownLatch(COUNT);

-    final ChannelInputStream[] inputStream = new ChannelInputStream[1];

-

-    getAcceptor().addListener(new IListener()

-    {

-      public void notifyEvent(IEvent event)

-      {

-        if (event instanceof IContainerEvent<?>)

-        {

-          @SuppressWarnings("unchecked")

-          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;

-          e.getDeltaElement().addListener(new IListener()

-          {

-            public void notifyEvent(IEvent event)

-            {

-              if (event instanceof IContainerEvent<?>)

-              {

-                @SuppressWarnings("unchecked")

-                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;

-                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)

-                {

-                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);

-                  counter.countDown();

-                }

-              }

-            }

-          });

-        }

-      }

-    });

-

-    IChannel channel = getConnector().openChannel();

-    assertEquals(true, counter.await(2, TimeUnit.SECONDS));

-    assertNotNull(inputStream[0]);

-

-    ChannelOutputStream outputStream = new ChannelOutputStream(channel);

-    PrintStream printer = new PrintStream(outputStream);

-    StringTokenizer tokenizer = HugeData.getTokenizer();

-    while (tokenizer.hasMoreTokens())

-    {

-      String token = tokenizer.nextToken();

-      printer.println(token);

-    }

-

-    outputStream.flushWithEOS();

-    outputStream.close();

-

-    try

-    {

-      InputStreamReader isr = new InputStreamReader(inputStream[0]);

-      BufferedReader reader = new BufferedReader(isr);

-      String line;

-      while ((line = reader.readLine()) != null)

-      {

-        msg(line);

-      }

-

-      isr.close();

-    }

-    catch (RuntimeException ex)

-    {

-      IOUtil.print(ex);

-    }

-  }

-

-  public void testTextStreamingDecoupled() throws Exception

-  {

-    final int COUNT = 1;

-    final CountDownLatch counter = new CountDownLatch(COUNT);

-    final ChannelInputStream[] inputStream = new ChannelInputStream[1];

-

-    getAcceptor().addListener(new IListener()

-    {

-      public void notifyEvent(IEvent event)

-      {

-        if (event instanceof IContainerEvent<?>)

-        {

-          @SuppressWarnings("unchecked")

-          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;

-          e.getDeltaElement().addListener(new IListener()

-          {

-            public void notifyEvent(IEvent event)

-            {

-              if (event instanceof IContainerEvent<?>)

-              {

-                @SuppressWarnings("unchecked")

-                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;

-                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)

-                {

-                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);

-                  counter.countDown();

-                }

-              }

-            }

-          });

-        }

-      }

-    });

-

-    final IConnector iConnector = getConnector();

-    final IChannel channel = iConnector.openChannel();

-    assertEquals(true, counter.await(2, TimeUnit.SECONDS));

-    assertNotNull(inputStream[0]);

-

-    new Thread()

-    {

-      @Override

-      public void run()

-      {

-        try

-        {

-          IBufferProvider bufferProvider = provideBufferProvider(iConnector);

-          ChannelOutputStream outputStream = new ChannelOutputStream(channel, bufferProvider);

-          PrintStream printer = new PrintStream(outputStream);

-          StringTokenizer tokenizer = HugeData.getTokenizer();

-          while (tokenizer.hasMoreTokens())

-          {

-            String token = tokenizer.nextToken();

-            printer.println(token);

-          }

-

-          outputStream.flushWithEOS();

-          outputStream.close();

-        }

-        catch (IOException ex)

-        {

-          IOUtil.print(ex);

-          fail(ex.getLocalizedMessage());

-        }

-      }

-    }.start();

-

-    try

-    {

-      InputStreamReader isr = new InputStreamReader(inputStream[0]);

-      BufferedReader reader = new BufferedReader(isr);

-      String line;

-      while ((line = reader.readLine()) != null)

-      {

-        msg(line);

-      }

-

-      isr.close();

-    }

-    catch (RuntimeException ex)

-    {

-      IOUtil.print(ex);

-    }

-  }

-

-  public void testDataStreaming() throws Exception

-  {

-    final int COUNT = 1;

-    final CountDownLatch counter = new CountDownLatch(COUNT);

-    final ChannelInputStream[] inputStream = new ChannelInputStream[1];

-

-    getAcceptor().addListener(new IListener()

-    {

-      public void notifyEvent(IEvent event)

-      {

-        if (event instanceof IContainerEvent<?>)

-        {

-          @SuppressWarnings("unchecked")

-          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;

-          e.getDeltaElement().addListener(new IListener()

-          {

-            public void notifyEvent(IEvent event)

-            {

-              if (event instanceof IContainerEvent<?>)

-              {

-                @SuppressWarnings("unchecked")

-                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;

-                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)

-                {

-                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);

-                  counter.countDown();

-                }

-              }

-            }

-          });

-        }

-      }

-    });

-

-    IChannel channel = getConnector().openChannel();

-    assertEquals(true, counter.await(2, TimeUnit.SECONDS));

-

-    ChannelOutputStream outputStream = new ChannelOutputStream(channel);

-    DataOutputStream dataOutput = new DataOutputStream(outputStream);

-    byte[] data = HugeData.getBytes();

-    dataOutput.writeInt(data.length);

-    dataOutput.write(data);

-    dataOutput.flush();

-    dataOutput.close();

-    outputStream.flush();

-

-    DataInputStream dataInput = new DataInputStream(inputStream[0]);

-    int size = dataInput.readInt();

-    byte[] b = new byte[size];

-    dataInput.read(b);

-    dataInput.close();

-

-    msg(new String(b));

-  }

-

-  /**

-   * @author Eike Stepper

-   */

-  public static final class TestProtocol extends Protocol<CountDownLatch>

-  {

-    public TestProtocol(CountDownLatch counter)

-    {

-      super(ServerFactory.TYPE);

-      setInfraStructure(counter);

-    }

-

-    public void handleBuffer(IBuffer buffer)

-    {

-      IOUtil.OUT().println("BUFFER ARRIVED"); //$NON-NLS-1$

-      buffer.release();

-      getInfraStructure().countDown();

-    }

-

-    /**

-     * @author Eike Stepper

-     */

-    public static class ServerFactory extends ServerProtocolFactory

-    {

-      public static final String TYPE = "test.protocol"; //$NON-NLS-1$

-

-      private CountDownLatch counter;

-

-      public ServerFactory(CountDownLatch counter)

-      {

-        super(TYPE);

-        this.counter = counter;

-      }

-

-      public TestProtocol create(String description) throws ProductCreationException

-      {

-        return new TestProtocol(counter);

-      }

-    }

-

-    /**

-     * @author Eike Stepper

-     */

-    public static class ClientFactory extends ClientProtocolFactory

-    {

-      public static final String TYPE = ServerFactory.TYPE;

-

-      public ClientFactory()

-      {

-        super(TYPE);

-      }

-

-      public TestProtocol create(String description) throws ProductCreationException

-      {

-        return new TestProtocol(null);

-      }

-    }

-  }

-

-  /**

-   * @author Eike Stepper

-   */

-  public static final class TCP extends TransportTest

-  {

-    @Override

-    protected boolean useJVMTransport()

-    {

-      return false;

-    }

-

-    @Override

-    protected boolean useSSLTransport()

-    {

-      return false;

-    }

-  }

-

-  /**

-   * @author Eike Stepper

-   */

-  public static final class JVM extends TransportTest

-  {

-    @Override

-    protected boolean useJVMTransport()

-    {

-      return true;

-    }

-

-    @Override

-    protected boolean useSSLTransport()

-    {

-      return false;

-    }

-  }

-

-  /**

-   * @author Teerawat Chaiyakijpichet (No Magic Asia Ltd.)

-   */

-  public static final class SSL extends TransportTest

-  {

-    @Override

-    protected boolean useJVMTransport()

-    {

-      return false;

-    }

-

-    @Override

-    protected boolean useSSLTransport()

-    {

-      return true;

-    }

-  }

-}

+/*
+ * Copyright (c) 2004 - 2012 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ *    Teerawat Chaiyakijpichet (No Magic Asia Ltd.) - SSL
+ */
+package org.eclipse.net4j.tests;
+
+import org.eclipse.net4j.Net4jUtil;
+import org.eclipse.net4j.buffer.IBuffer;
+import org.eclipse.net4j.buffer.IBufferProvider;
+import org.eclipse.net4j.channel.ChannelInputStream;
+import org.eclipse.net4j.channel.ChannelOutputStream;
+import org.eclipse.net4j.channel.IChannel;
+import org.eclipse.net4j.connector.IConnector;
+import org.eclipse.net4j.tests.data.HugeData;
+import org.eclipse.net4j.util.container.IContainerDelta;
+import org.eclipse.net4j.util.container.IContainerEvent;
+import org.eclipse.net4j.util.event.IEvent;
+import org.eclipse.net4j.util.event.IListener;
+import org.eclipse.net4j.util.factory.IFactory;
+import org.eclipse.net4j.util.factory.ProductCreationException;
+import org.eclipse.net4j.util.io.IOUtil;
+
+import org.eclipse.spi.net4j.ClientProtocolFactory;
+import org.eclipse.spi.net4j.Connector;
+import org.eclipse.spi.net4j.Protocol;
+import org.eclipse.spi.net4j.ServerProtocolFactory;
+
+import java.io.BufferedReader;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.PrintStream;
+import java.nio.ByteBuffer;
+import java.util.StringTokenizer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author Eike Stepper
+ */
+public abstract class TransportTest extends AbstractProtocolTest
+{
+  public TransportTest()
+  {
+  }
+
+  @Override
+  protected abstract boolean useJVMTransport();
+
+  protected IBuffer provideBuffer()
+  {
+    return provideBuffer(null);
+  }
+
+  protected IBuffer provideBuffer(IConnector iConnector)
+  {
+    IBuffer buffer = null;
+    if (!useJVMTransport() && useSSLTransport())
+    {
+      // cannot use buffer provider from net4j need to use SSL Buffer inside the SSLConnector.
+      buffer = ((Connector)iConnector).provideBuffer();
+    }
+    else
+    {
+      IBufferProvider bufferProvider = Net4jUtil.getBufferProvider(container);
+      buffer = bufferProvider.provideBuffer();
+    }
+
+    return buffer;
+  }
+
+  private void registerClientFactory(IFactory factory)
+  {
+    if (!useJVMTransport() && useSSLTransport())
+    {
+      // need separate container between client and server for SSL.
+      separateContainer.registerFactory(factory);
+    }
+    else
+    {
+      container.registerFactory(factory);
+    }
+  }
+
+  protected IBufferProvider provideBufferProvider(IConnector iConnector)
+  {
+    IBufferProvider bufferProvider = null;
+    if (!useJVMTransport() && useSSLTransport())
+    {
+      // cannot use buffer provider from net4j need to use SSL Buffer inside the SSLConnector.
+      bufferProvider = ((Connector)iConnector).getConfig().getBufferProvider();
+    }
+    else
+    {
+      bufferProvider = Net4jUtil.getBufferProvider(container);
+    }
+
+    return bufferProvider;
+  }
+
+  public void testConnect() throws Exception
+  {
+    startTransport();
+  }
+
+  public void testSendBuffer() throws Exception
+  {
+    startTransport();
+    IConnector connecter = getConnector();
+    IChannel channel = connecter.openChannel();
+    for (int i = 0; i < 3; i++)
+    {
+      IBuffer buffer = provideBuffer(connecter);
+
+      ByteBuffer byteBuffer = buffer.startPutting(channel.getID());
+      byteBuffer.putInt(1970);
+      channel.sendBuffer(buffer);
+    }
+  }
+
+  public void testSendEmptyBuffer() throws Exception
+  {
+    startTransport();
+    IConnector connecter = getConnector();
+    IChannel channel = connecter.openChannel();
+    for (int i = 0; i < 3; i++)
+    {
+      IBuffer buffer = provideBuffer(connecter);
+      buffer.startPutting(channel.getID());
+      channel.sendBuffer(buffer);
+    }
+  }
+
+  public void testSendEmptyBuffer2() throws Exception
+  {
+    startTransport();
+    IConnector connecter = getConnector();
+    IChannel channel = connecter.openChannel();
+    for (int i = 0; i < 3; i++)
+    {
+      IBuffer buffer = provideBuffer(connecter);
+      channel.sendBuffer(buffer);
+    }
+  }
+
+  public void testHandleBuffer() throws Exception
+  {
+    final int COUNT = 3;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    container.registerFactory(new TestProtocol.ServerFactory(counter));
+    // need to handle about separating container between client and server for SSL.
+    registerClientFactory(new TestProtocol.ClientFactory());
+    startTransport();
+    IConnector iConnecter = getConnector();
+    IChannel channel = iConnecter.openChannel(TestProtocol.ClientFactory.TYPE, null);
+    for (int i = 0; i < COUNT; i++)
+    {
+      IBuffer buffer = provideBuffer(iConnecter);
+      ByteBuffer byteBuffer = buffer.startPutting(channel.getID());
+      byteBuffer.putInt(1970);
+      channel.sendBuffer(buffer);
+      sleep(50);
+    }
+
+    assertEquals(true, counter.await(2, TimeUnit.SECONDS));
+  }
+
+  public void testHandleEmptyBuffer() throws Exception
+  {
+    final int COUNT = 3;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    container.registerFactory(new TestProtocol.ServerFactory(counter));
+    // need to handle about separating container between client and server for SSL.
+    registerClientFactory(new TestProtocol.ClientFactory());
+
+    startTransport();
+    IConnector connecter = getConnector();
+    IChannel channel = connecter.openChannel(TestProtocol.ClientFactory.TYPE, null);
+    for (int i = 0; i < COUNT; i++)
+    {
+      IBuffer buffer = provideBuffer(connecter);
+      buffer.startPutting(channel.getID());
+      channel.sendBuffer(buffer);
+      sleep(50);
+    }
+
+    assertEquals(COUNT, counter.getCount());
+  }
+
+  public void testHandleEmptyBuffer2() throws Exception
+  {
+    final int COUNT = 3;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    container.registerFactory(new TestProtocol.ServerFactory(counter));
+    // need to handle about separating container between client and server for SSL.
+    registerClientFactory(new TestProtocol.ClientFactory());
+
+    startTransport();
+    IConnector connecter = getConnector();
+    IChannel channel = connecter.openChannel(TestProtocol.ClientFactory.TYPE, null);
+    for (int i = 0; i < COUNT; i++)
+    {
+      IBuffer buffer = provideBuffer(connecter);
+      channel.sendBuffer(buffer);
+      sleep(50);
+    }
+
+    assertEquals(COUNT, counter.getCount());
+  }
+
+  public void testStreaming() throws Exception
+  {
+    final int COUNT = 1;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    final ChannelInputStream[] inputStream = new ChannelInputStream[1];
+
+    getAcceptor().addListener(new IListener()
+    {
+      public void notifyEvent(IEvent event)
+      {
+        if (event instanceof IContainerEvent<?>)
+        {
+          @SuppressWarnings("unchecked")
+          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;
+          e.getDeltaElement().addListener(new IListener()
+          {
+            public void notifyEvent(IEvent event)
+            {
+              if (event instanceof IContainerEvent<?>)
+              {
+                @SuppressWarnings("unchecked")
+                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;
+                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)
+                {
+                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);
+                  counter.countDown();
+                }
+              }
+            }
+          });
+        }
+      }
+    });
+
+    IChannel channel = getConnector().openChannel();
+    assertEquals(true, counter.await(2, TimeUnit.SECONDS));
+    assertNotNull(inputStream[0]);
+
+    ChannelOutputStream outputStream = new ChannelOutputStream(channel);
+    outputStream.write(HugeData.getBytes());
+    outputStream.flushWithEOS();
+    outputStream.close();
+
+    try
+    {
+      InputStreamReader isr = new InputStreamReader(inputStream[0]);
+      BufferedReader reader = new BufferedReader(isr);
+      String line;
+      while ((line = reader.readLine()) != null)
+      {
+        msg(line);
+      }
+
+      isr.close();
+    }
+    catch (RuntimeException ex)
+    {
+      IOUtil.print(ex);
+    }
+  }
+
+  /**
+   * TODO Fails occasionally ;-( Caused by: java.lang.IllegalStateException: selectionKey == null
+   */
+  public void testTextStreaming() throws Exception
+  {
+    final int COUNT = 1;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    final ChannelInputStream[] inputStream = new ChannelInputStream[1];
+
+    getAcceptor().addListener(new IListener()
+    {
+      public void notifyEvent(IEvent event)
+      {
+        if (event instanceof IContainerEvent<?>)
+        {
+          @SuppressWarnings("unchecked")
+          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;
+          e.getDeltaElement().addListener(new IListener()
+          {
+            public void notifyEvent(IEvent event)
+            {
+              if (event instanceof IContainerEvent<?>)
+              {
+                @SuppressWarnings("unchecked")
+                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;
+                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)
+                {
+                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);
+                  counter.countDown();
+                }
+              }
+            }
+          });
+        }
+      }
+    });
+
+    IChannel channel = getConnector().openChannel();
+    assertEquals(true, counter.await(2, TimeUnit.SECONDS));
+    assertNotNull(inputStream[0]);
+
+    ChannelOutputStream outputStream = new ChannelOutputStream(channel);
+    PrintStream printer = new PrintStream(outputStream);
+    StringTokenizer tokenizer = HugeData.getTokenizer();
+    while (tokenizer.hasMoreTokens())
+    {
+      String token = tokenizer.nextToken();
+      printer.println(token);
+    }
+
+    outputStream.flushWithEOS();
+    outputStream.close();
+
+    try
+    {
+      InputStreamReader isr = new InputStreamReader(inputStream[0]);
+      BufferedReader reader = new BufferedReader(isr);
+      String line;
+      while ((line = reader.readLine()) != null)
+      {
+        msg(line);
+      }
+
+      isr.close();
+    }
+    catch (RuntimeException ex)
+    {
+      IOUtil.print(ex);
+    }
+  }
+
+  public void testTextStreamingDecoupled() throws Exception
+  {
+    final int COUNT = 1;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    final ChannelInputStream[] inputStream = new ChannelInputStream[1];
+
+    getAcceptor().addListener(new IListener()
+    {
+      public void notifyEvent(IEvent event)
+      {
+        if (event instanceof IContainerEvent<?>)
+        {
+          @SuppressWarnings("unchecked")
+          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;
+          e.getDeltaElement().addListener(new IListener()
+          {
+            public void notifyEvent(IEvent event)
+            {
+              if (event instanceof IContainerEvent<?>)
+              {
+                @SuppressWarnings("unchecked")
+                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;
+                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)
+                {
+                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);
+                  counter.countDown();
+                }
+              }
+            }
+          });
+        }
+      }
+    });
+
+    final IConnector iConnector = getConnector();
+    final IChannel channel = iConnector.openChannel();
+    assertEquals(true, counter.await(2, TimeUnit.SECONDS));
+    assertNotNull(inputStream[0]);
+
+    new Thread()
+    {
+      @Override
+      public void run()
+      {
+        try
+        {
+          IBufferProvider bufferProvider = provideBufferProvider(iConnector);
+          ChannelOutputStream outputStream = new ChannelOutputStream(channel, bufferProvider);
+          PrintStream printer = new PrintStream(outputStream);
+          StringTokenizer tokenizer = HugeData.getTokenizer();
+          while (tokenizer.hasMoreTokens())
+          {
+            String token = tokenizer.nextToken();
+            printer.println(token);
+          }
+
+          outputStream.flushWithEOS();
+          outputStream.close();
+        }
+        catch (IOException ex)
+        {
+          IOUtil.print(ex);
+          fail(ex.getLocalizedMessage());
+        }
+      }
+    }.start();
+
+    try
+    {
+      InputStreamReader isr = new InputStreamReader(inputStream[0]);
+      BufferedReader reader = new BufferedReader(isr);
+      String line;
+      while ((line = reader.readLine()) != null)
+      {
+        msg(line);
+      }
+
+      isr.close();
+    }
+    catch (RuntimeException ex)
+    {
+      IOUtil.print(ex);
+    }
+  }
+
+  public void testDataStreaming() throws Exception
+  {
+    final int COUNT = 1;
+    final CountDownLatch counter = new CountDownLatch(COUNT);
+    final ChannelInputStream[] inputStream = new ChannelInputStream[1];
+
+    getAcceptor().addListener(new IListener()
+    {
+      public void notifyEvent(IEvent event)
+      {
+        if (event instanceof IContainerEvent<?>)
+        {
+          @SuppressWarnings("unchecked")
+          IContainerEvent<IConnector> e = (IContainerEvent<IConnector>)event;
+          e.getDeltaElement().addListener(new IListener()
+          {
+            public void notifyEvent(IEvent event)
+            {
+              if (event instanceof IContainerEvent<?>)
+              {
+                @SuppressWarnings("unchecked")
+                IContainerEvent<IChannel> e = (IContainerEvent<IChannel>)event;
+                if (e.getDeltaKind() == IContainerDelta.Kind.ADDED)
+                {
+                  inputStream[0] = new ChannelInputStream(e.getDeltaElement(), 2000);
+                  counter.countDown();
+                }
+              }
+            }
+          });
+        }
+      }
+    });
+
+    IChannel channel = getConnector().openChannel();
+    assertEquals(true, counter.await(2, TimeUnit.SECONDS));
+
+    ChannelOutputStream outputStream = new ChannelOutputStream(channel);
+    DataOutputStream dataOutput = new DataOutputStream(outputStream);
+    byte[] data = HugeData.getBytes();
+    dataOutput.writeInt(data.length);
+    dataOutput.write(data);
+    dataOutput.flush();
+    dataOutput.close();
+    outputStream.flush();
+
+    DataInputStream dataInput = new DataInputStream(inputStream[0]);
+    int size = dataInput.readInt();
+    byte[] b = new byte[size];
+    dataInput.read(b);
+    dataInput.close();
+
+    msg(new String(b));
+  }
+
+  /**
+   * @author Eike Stepper
+   */
+  public static final class TestProtocol extends Protocol<CountDownLatch>
+  {
+    public TestProtocol(CountDownLatch counter)
+    {
+      super(ServerFactory.TYPE);
+      setInfraStructure(counter);
+    }
+
+    public void handleBuffer(IBuffer buffer)
+    {
+      IOUtil.OUT().println("BUFFER ARRIVED"); //$NON-NLS-1$
+      buffer.release();
+      getInfraStructure().countDown();
+    }
+
+    /**
+     * @author Eike Stepper
+     */
+    public static class ServerFactory extends ServerProtocolFactory
+    {
+      public static final String TYPE = "test.protocol"; //$NON-NLS-1$
+
+      private CountDownLatch counter;
+
+      public ServerFactory(CountDownLatch counter)
+      {
+        super(TYPE);
+        this.counter = counter;
+      }
+
+      public TestProtocol create(String description) throws ProductCreationException
+      {
+        return new TestProtocol(counter);
+      }
+    }
+
+    /**
+     * @author Eike Stepper
+     */
+    public static class ClientFactory extends ClientProtocolFactory
+    {
+      public static final String TYPE = ServerFactory.TYPE;
+
+      public ClientFactory()
+      {
+        super(TYPE);
+      }
+
+      public TestProtocol create(String description) throws ProductCreationException
+      {
+        return new TestProtocol(null);
+      }
+    }
+  }
+
+  /**
+   * @author Eike Stepper
+   */
+  public static final class TCP extends TransportTest
+  {
+    @Override
+    protected boolean useJVMTransport()
+    {
+      return false;
+    }
+
+    @Override
+    protected boolean useSSLTransport()
+    {
+      return false;
+    }
+  }
+
+  /**
+   * @author Eike Stepper
+   */
+  public static final class JVM extends TransportTest
+  {
+    @Override
+    protected boolean useJVMTransport()
+    {
+      return true;
+    }
+
+    @Override
+    protected boolean useSSLTransport()
+    {
+      return false;
+    }
+  }
+
+  /**
+   * @author Teerawat Chaiyakijpichet (No Magic Asia Ltd.)
+   */
+  public static final class SSL extends TransportTest
+  {
+    @Override
+    protected boolean useJVMTransport()
+    {
+      return false;
+    }
+
+    @Override
+    protected boolean useSSLTransport()
+    {
+      return true;
+    }
+  }
+}
diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/buffer/Buffer.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/buffer/Buffer.java
index 69123bc6fe..815068a401 100644
--- a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/buffer/Buffer.java
+++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/buffer/Buffer.java
@@ -1,432 +1,438 @@
-/*

- * Copyright (c) 2004 - 2012 Eike Stepper (Berlin, Germany) and others.

- * All rights reserved. This program and the accompanying materials

- * are made available under the terms of the Eclipse Public License v1.0

- * which accompanies this distribution, and is available at

- * http://www.eclipse.org/legal/epl-v10.html

- *

- * Contributors:

- *    Eike Stepper - initial API and implementation

- *    Caspar De Groot - maintenance

- */

-package org.eclipse.internal.net4j.buffer;

-

-import org.eclipse.net4j.buffer.BufferState;

-import org.eclipse.net4j.buffer.IBuffer;

-import org.eclipse.net4j.buffer.IBufferProvider;

-import org.eclipse.net4j.util.HexUtil;

-import org.eclipse.net4j.util.IErrorHandler;

-import org.eclipse.net4j.util.ReflectUtil;

-import org.eclipse.net4j.util.StringUtil;

-import org.eclipse.net4j.util.om.trace.ContextTracer;

-

-import org.eclipse.internal.net4j.bundle.OM;

-

-import org.eclipse.spi.net4j.InternalBuffer;

-

-import java.io.IOException;

-import java.nio.ByteBuffer;

-import java.nio.channels.ClosedChannelException;

-import java.nio.channels.SocketChannel;

-import java.text.MessageFormat;

-

-/**

- * @author Eike Stepper

- * @since 4.0

- */

-public class Buffer implements InternalBuffer

-{

-  public static final int EOS_OFFSET = 1;

-

-  private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG_BUFFER, Buffer.class);

-

-  private IErrorHandler errorHandler;

-

-  private IBufferProvider bufferProvider;

-

-  private short channelID;

-

-  private boolean eos;

-

-  private BufferState state = BufferState.INITIAL;

-

-  private ByteBuffer byteBuffer;

-

-  public Buffer(IBufferProvider provider, short capacity)

-  {

-    bufferProvider = provider;

-    byteBuffer = ByteBuffer.allocateDirect(capacity);

-  }

-

-  public boolean isEOS()

-  {

-    return eos;

-  }

-

-  public void setEOS(boolean eos)

-  {

-    this.eos = eos;

-  }

-

-  public IBufferProvider getBufferProvider()

-  {

-    return bufferProvider;

-  }

-

-  public void setBufferProvider(IBufferProvider bufferProvider)

-  {

-    this.bufferProvider = bufferProvider;

-  }

-

-  public short getChannelID()

-  {

-    if (state == BufferState.INITIAL || state == BufferState.READING_HEADER)

-    {

-      throw new IllegalStateException(toString());

-    }

-

-    return channelID;

-  }

-

-  public void setChannelID(short channelID)

-  {

-    this.channelID = channelID;

-  }

-

-  public short getCapacity()

-  {

-    return (short)byteBuffer.capacity();

-  }

-

-  public BufferState getState()

-  {

-    return state;

-  }

-

-  public void setState(BufferState state)

-  {

-    this.state = state;

-  }

-

-  public ByteBuffer getByteBuffer()

-  {

-    return byteBuffer;

-  }

-

-  public void setByteBuffer(ByteBuffer buffer)

-  {

-    byteBuffer = buffer;

-  }

-

-  public void clear()

-  {

-    state = BufferState.INITIAL;

-    channelID = NO_CHANNEL;

-    eos = false;

-    byteBuffer.clear();

-  }

-

-  public void release()

-  {

-    if (state != BufferState.RELEASED)

-    {

-      state = BufferState.RELEASED;

-      errorHandler = null;

-      if (bufferProvider != null)

-      {

-        bufferProvider.retainBuffer(this);

-      }

-    }

-  }

-

-  public void dispose()

-  {

-    state = BufferState.DISPOSED;

-    bufferProvider = null;

-    byteBuffer = null;

-  }

-

-  public ByteBuffer startGetting(SocketChannel socketChannel) throws IOException

-  {

-    try

-    {

-      if (state != BufferState.INITIAL && state != BufferState.READING_HEADER && state != BufferState.READING_BODY)

-      {

-        throw new IllegalStateException(toString());

-      }

-

-      if (state == BufferState.INITIAL)

-      {

-        byteBuffer.limit(IBuffer.HEADER_SIZE);

-        state = BufferState.READING_HEADER;

-      }

-

-      if (state == BufferState.READING_HEADER)

-      {

-        readChannel(socketChannel, byteBuffer);

-        if (byteBuffer.hasRemaining())

-        {

-          return null;

-        }

-

-        byteBuffer.flip();

-        channelID = byteBuffer.getShort();

-        short payloadSize = byteBuffer.getShort();

-        if (payloadSize < 0)

-        {

-          eos = true;

-          payloadSize = (short)-payloadSize;

-        }

-

-        payloadSize -= EOS_OFFSET;

-

-        byteBuffer.clear();

-        byteBuffer.limit(payloadSize);

-        state = BufferState.READING_BODY;

-      }

-

-      readChannel(socketChannel, byteBuffer);

-      if (byteBuffer.hasRemaining())

-      {

-        return null;

-      }

-

-      if (TRACER.isEnabled())

-      {

-        TRACER.trace("Read " + byteBuffer.limit() + " bytes" //$NON-NLS-1$ //$NON-NLS-2$

-            + (eos ? " (EOS)" : "") + StringUtil.NL + formatContent(false)); //$NON-NLS-1$ //$NON-NLS-2$

-      }

-

-      byteBuffer.flip();

-      state = BufferState.GETTING;

-      return byteBuffer;

-    }

-    catch (IOException ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-    catch (RuntimeException ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-    catch (Error ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-  }

-

-  public ByteBuffer startPutting(short channelID)

-  {

-    try

-    {

-      if (state == BufferState.PUTTING)

-      {

-        if (channelID != this.channelID)

-        {

-          throw new IllegalArgumentException("channelID != this.channelID"); //$NON-NLS-1$

-        }

-      }

-      else if (state != BufferState.INITIAL)

-      {

-        throw new IllegalStateException("state: " + state); //$NON-NLS-1$

-      }

-      else

-      {

-        state = BufferState.PUTTING;

-        this.channelID = channelID;

-

-        byteBuffer.clear();

-        byteBuffer.position(IBuffer.HEADER_SIZE);

-      }

-

-      return byteBuffer;

-    }

-    catch (RuntimeException ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-    catch (Error ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-  }

-

-  /**

-   * @return <code>true</code> if the buffer has been completely written, <code>false</code> otherwise.

-   */

-  public boolean write(SocketChannel socketChannel) throws IOException

-  {

-    try

-    {

-      if (state != BufferState.PUTTING && state != BufferState.WRITING)

-      {

-        throw new IllegalStateException(toString());

-      }

-

-      if (state == BufferState.PUTTING)

-      {

-        if (channelID == NO_CHANNEL)

-        {

-          throw new IllegalStateException("channelID == NO_CHANNEL"); //$NON-NLS-1$

-        }

-

-        int payloadSize = byteBuffer.position() - IBuffer.HEADER_SIZE + EOS_OFFSET;

-        if (eos)

-        {

-          payloadSize = -payloadSize;

-        }

-

-        if (TRACER.isEnabled())

-        {

-          TRACER.trace("Writing " + (Math.abs(payloadSize) - 1) + " bytes" //$NON-NLS-1$ //$NON-NLS-2$

-              + (eos ? " (EOS)" : "") + StringUtil.NL + formatContent(false)); //$NON-NLS-1$ //$NON-NLS-2$

-        }

-

-        byteBuffer.flip();

-        byteBuffer.putShort(channelID);

-        byteBuffer.putShort((short)payloadSize);

-        byteBuffer.position(0);

-        state = BufferState.WRITING;

-      }

-

-      int numBytes = socketChannel.write(byteBuffer);

-      if (numBytes == -1)

-      {

-        throw new IOException("Channel closed"); //$NON-NLS-1$

-      }

-

-      if (byteBuffer.hasRemaining())

-      {

-        return false;

-      }

-

-      clear();

-      return true;

-    }

-    catch (IOException ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-    catch (RuntimeException ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-    catch (Error ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-  }

-

-  public void flip()

-  {

-    try

-    {

-      if (state != BufferState.PUTTING)

-      {

-        throw new IllegalStateException(toString());

-      }

-

-      byteBuffer.flip();

-      byteBuffer.position(IBuffer.HEADER_SIZE);

-      state = BufferState.GETTING;

-    }

-    catch (RuntimeException ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-    catch (Error ex)

-    {

-      handleError(ex);

-      throw ex;

-    }

-  }

-

-  @Override

-  public String toString()

-  {

-    return MessageFormat.format("Buffer@{0}[{1}]", ReflectUtil.getID(this), state); //$NON-NLS-1$

-  }

-

-  @SuppressWarnings("deprecation")

-  public String formatContent(boolean showHeader)

-  {

-    final int oldPosition = byteBuffer.position();

-    final int oldLimit = byteBuffer.limit();

-

-    try

-    {

-      if (state != BufferState.GETTING)

-      {

-        byteBuffer.flip();

-      }

-

-      if (state == BufferState.PUTTING && !showHeader)

-      {

-        byteBuffer.position(IBuffer.HEADER_SIZE);

-      }

-

-      StringBuilder builder = new StringBuilder();

-      while (byteBuffer.hasRemaining())

-      {

-        byte b = byteBuffer.get();

-        HexUtil.appendHex(builder, b < 0 ? ~b : b);

-        builder.append(' ');

-      }

-

-      return builder.toString();

-    }

-    finally

-    {

-      byteBuffer.position(oldPosition);

-      byteBuffer.limit(oldLimit);

-    }

-  }

-

-  public IErrorHandler getErrorHandler()

-  {

-    return errorHandler;

-  }

-

-  public void setErrorHandler(IErrorHandler errorHandler)

-  {

-    this.errorHandler = errorHandler;

-  }

-

-  public void handleError(Throwable t)

-  {

-    if (errorHandler != null)

-    {

-      errorHandler.handleError(t);

-    }

-

-    release();

-  }

-

-  private static void readChannel(SocketChannel socketChannel, ByteBuffer buffer) throws ClosedChannelException

-  {

-    try

-    {

-      if (socketChannel.read(buffer) == -1)

-      {

-        throw new ClosedChannelException();

-      }

-    }

-    catch (ClosedChannelException ex)

-    {

-      throw ex;

-    }

-    catch (IOException ex)

-    {

-      throw new ClosedChannelException();

-    }

-  }

-}

+/*
+ * Copyright (c) 2004 - 2012 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ *    Caspar De Groot - maintenance
+ */
+package org.eclipse.internal.net4j.buffer;
+
+import org.eclipse.net4j.buffer.BufferState;
+import org.eclipse.net4j.buffer.IBuffer;
+import org.eclipse.net4j.buffer.IBufferProvider;
+import org.eclipse.net4j.util.HexUtil;
+import org.eclipse.net4j.util.IErrorHandler;
+import org.eclipse.net4j.util.ReflectUtil;
+import org.eclipse.net4j.util.StringUtil;
+import org.eclipse.net4j.util.om.trace.ContextTracer;
+
+import org.eclipse.internal.net4j.bundle.OM;
+
+import org.eclipse.spi.net4j.InternalBuffer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.SocketChannel;
+import java.text.MessageFormat;
+
+/**
+ * @author Eike Stepper
+ * @since 4.0
+ */
+public class Buffer implements InternalBuffer
+{
+  public static final int EOS_OFFSET = 1;
+
+  private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG_BUFFER, Buffer.class);
+
+  private IErrorHandler errorHandler;
+
+  private IBufferProvider bufferProvider;
+
+  private short channelID;
+
+  private boolean eos;
+
+  private BufferState state = BufferState.INITIAL;
+
+  private ByteBuffer byteBuffer;
+
+  public Buffer(IBufferProvider provider, short capacity)
+  {
+    bufferProvider = provider;
+    byteBuffer = ByteBuffer.allocateDirect(capacity);
+  }
+
+  public boolean isEOS()
+  {
+    return eos;
+  }
+
+  public void setEOS(boolean eos)
+  {
+    this.eos = eos;
+  }
+
+  public IBufferProvider getBufferProvider()
+  {
+    return bufferProvider;
+  }
+
+  public void setBufferProvider(IBufferProvider bufferProvider)
+  {
+    this.bufferProvider = bufferProvider;
+  }
+
+  public short getChannelID()
+  {
+    if (state == BufferState.INITIAL || state == BufferState.READING_HEADER)
+    {
+      throw new IllegalStateException(toString());
+    }
+
+    return channelID;
+  }
+
+  public void setChannelID(short channelID)
+  {
+    this.channelID = channelID;
+  }
+
+  public short getCapacity()
+  {
+    return (short)byteBuffer.capacity();
+  }
+
+  public BufferState getState()
+  {
+    return state;
+  }
+
+  public void setState(BufferState state)
+  {
+    this.state = state;
+  }
+
+  public ByteBuffer getByteBuffer()
+  {
+    return byteBuffer;
+  }
+
+  public void setByteBuffer(ByteBuffer buffer)
+  {
+    byteBuffer = buffer;
+  }
+
+  public void clear()
+  {
+    state = BufferState.INITIAL;
+    channelID = NO_CHANNEL;
+    eos = false;
+    byteBuffer.clear();
+  }
+
+  public void release()
+  {
+    if (state != BufferState.RELEASED)
+    {
+      state = BufferState.RELEASED;
+      errorHandler = null;
+      if (bufferProvider != null)
+      {
+        bufferProvider.retainBuffer(this);
+      }
+    }
+  }
+
+  public void dispose()
+  {
+    state = BufferState.DISPOSED;
+    bufferProvider = null;
+    byteBuffer = null;
+  }
+
+  public ByteBuffer startGetting(SocketChannel socketChannel) throws IOException
+  {
+    try
+    {
+      if (state != BufferState.INITIAL && state != BufferState.READING_HEADER && state != BufferState.READING_BODY)
+      {
+        throw new IllegalStateException(toString());
+      }
+
+      if (state == BufferState.INITIAL)
+      {
+        byteBuffer.limit(IBuffer.HEADER_SIZE);
+        state = BufferState.READING_HEADER;
+      }
+
+      if (state == BufferState.READING_HEADER)
+      {
+        readChannel(socketChannel, byteBuffer);
+        if (byteBuffer.hasRemaining())
+        {
+          return null;
+        }
+
+        byteBuffer.flip();
+        channelID = byteBuffer.getShort();
+        short payloadSize = byteBuffer.getShort();
+        if (payloadSize < 0)
+        {
+          eos = true;
+          payloadSize = (short)-payloadSize;
+        }
+
+        payloadSize -= EOS_OFFSET;
+
+        byteBuffer.clear();
+        byteBuffer.limit(payloadSize);
+        state = BufferState.READING_BODY;
+      }
+
+      readChannel(socketChannel, byteBuffer);
+      if (byteBuffer.hasRemaining())
+      {
+        return null;
+      }
+
+      if (TRACER.isEnabled())
+      {
+        TRACER.trace("Read " + byteBuffer.limit() + " bytes" //$NON-NLS-1$ //$NON-NLS-2$
+            + (eos ? " (EOS)" : "") + StringUtil.NL + formatContent(false)); //$NON-NLS-1$ //$NON-NLS-2$
+      }
+
+      byteBuffer.flip();
+      state = BufferState.GETTING;
+      return byteBuffer;
+    }
+    catch (IOException ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+    catch (RuntimeException ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+    catch (Error ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+  }
+
+  public ByteBuffer startPutting(short channelID)
+  {
+    try
+    {
+      if (state == BufferState.PUTTING)
+      {
+        if (channelID != this.channelID)
+        {
+          throw new IllegalArgumentException("channelID != this.channelID"); //$NON-NLS-1$
+        }
+      }
+      else if (state != BufferState.INITIAL)
+      {
+        throw new IllegalStateException("state: " + state); //$NON-NLS-1$
+      }
+      else
+      {
+        state = BufferState.PUTTING;
+        this.channelID = channelID;
+
+        byteBuffer.clear();
+        byteBuffer.position(IBuffer.HEADER_SIZE);
+      }
+
+      return byteBuffer;
+    }
+    catch (RuntimeException ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+    catch (Error ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+  }
+
+  /**
+   * @return <code>true</code> if the buffer has been completely written, <code>false</code> otherwise.
+   */
+  public boolean write(SocketChannel socketChannel) throws IOException
+  {
+    try
+    {
+      if (byteBuffer.position() == HEADER_SIZE)
+      {
+        clear();
+        return true; // *Pretend* that this empty buffer has been written
+      }
+
+      if (state != BufferState.PUTTING && state != BufferState.WRITING)
+      {
+        throw new IllegalStateException(toString());
+      }
+
+      if (state == BufferState.PUTTING)
+      {
+        if (channelID == NO_CHANNEL)
+        {
+          throw new IllegalStateException("channelID == NO_CHANNEL"); //$NON-NLS-1$
+        }
+
+        int payloadSize = byteBuffer.position() - IBuffer.HEADER_SIZE + EOS_OFFSET;
+        if (eos)
+        {
+          payloadSize = -payloadSize;
+        }
+
+        if (TRACER.isEnabled())
+        {
+          TRACER.trace("Writing " + (Math.abs(payloadSize) - 1) + " bytes" //$NON-NLS-1$ //$NON-NLS-2$
+              + (eos ? " (EOS)" : "") + StringUtil.NL + formatContent(false)); //$NON-NLS-1$ //$NON-NLS-2$
+        }
+
+        byteBuffer.flip();
+        byteBuffer.putShort(channelID);
+        byteBuffer.putShort((short)payloadSize);
+        byteBuffer.position(0);
+        state = BufferState.WRITING;
+      }
+
+      int numBytes = socketChannel.write(byteBuffer);
+      if (numBytes == -1)
+      {
+        throw new IOException("Channel closed"); //$NON-NLS-1$
+      }
+
+      if (byteBuffer.hasRemaining())
+      {
+        return false;
+      }
+
+      clear();
+      return true;
+    }
+    catch (IOException ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+    catch (RuntimeException ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+    catch (Error ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+  }
+
+  public void flip()
+  {
+    try
+    {
+      if (state != BufferState.PUTTING)
+      {
+        throw new IllegalStateException(toString());
+      }
+
+      byteBuffer.flip();
+      byteBuffer.position(IBuffer.HEADER_SIZE);
+      state = BufferState.GETTING;
+    }
+    catch (RuntimeException ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+    catch (Error ex)
+    {
+      handleError(ex);
+      throw ex;
+    }
+  }
+
+  @Override
+  public String toString()
+  {
+    return MessageFormat.format("Buffer@{0}[{1}]", ReflectUtil.getID(this), state); //$NON-NLS-1$
+  }
+
+  @SuppressWarnings("deprecation")
+  public String formatContent(boolean showHeader)
+  {
+    final int oldPosition = byteBuffer.position();
+    final int oldLimit = byteBuffer.limit();
+
+    try
+    {
+      if (state != BufferState.GETTING)
+      {
+        byteBuffer.flip();
+      }
+
+      if (state == BufferState.PUTTING && !showHeader)
+      {
+        byteBuffer.position(IBuffer.HEADER_SIZE);
+      }
+
+      StringBuilder builder = new StringBuilder();
+      while (byteBuffer.hasRemaining())
+      {
+        byte b = byteBuffer.get();
+        HexUtil.appendHex(builder, b < 0 ? ~b : b);
+        builder.append(' ');
+      }
+
+      return builder.toString();
+    }
+    finally
+    {
+      byteBuffer.position(oldPosition);
+      byteBuffer.limit(oldLimit);
+    }
+  }
+
+  public IErrorHandler getErrorHandler()
+  {
+    return errorHandler;
+  }
+
+  public void setErrorHandler(IErrorHandler errorHandler)
+  {
+    this.errorHandler = errorHandler;
+  }
+
+  public void handleError(Throwable t)
+  {
+    if (errorHandler != null)
+    {
+      errorHandler.handleError(t);
+    }
+
+    release();
+  }
+
+  private static void readChannel(SocketChannel socketChannel, ByteBuffer buffer) throws ClosedChannelException
+  {
+    try
+    {
+      if (socketChannel.read(buffer) == -1)
+      {
+        throw new ClosedChannelException();
+      }
+    }
+    catch (ClosedChannelException ex)
+    {
+      throw ex;
+    }
+    catch (IOException ex)
+    {
+      throw new ClosedChannelException();
+    }
+  }
+}