Skip to main content
aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorEike Stepper2012-04-18 16:14:02 +0000
committerEike Stepper2012-04-18 16:14:02 +0000
commit3e7b56f4756edd8e9bcf430c49ad4b684ba6aa90 (patch)
treeaf0d8d483b0555b3abb416ee195152a3cfa7609f /plugins
parent604d4fb599ca3c0faba611e7fb4ec6f993ed97a4 (diff)
downloadcdo-3e7b56f4756edd8e9bcf430c49ad4b684ba6aa90.tar.gz
cdo-3e7b56f4756edd8e9bcf430c49ad4b684ba6aa90.tar.xz
cdo-3e7b56f4756edd8e9bcf430c49ad4b684ba6aa90.zip
[375939] Payload size of 21331 cause Buffer error when CDO UI is used
https://bugs.eclipse.org/bugs/show_bug.cgi?id=375939
Diffstat (limited to 'plugins')
-rw-r--r--plugins/org.eclipse.net4j.jvm/src/org/eclipse/net4j/internal/jvm/JVMConnector.java328
-rw-r--r--plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/TransportTest.java1136
-rw-r--r--plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/buffer/Buffer.java870
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();
+ }
+ }
+}

Back to the top