From 3e7b56f4756edd8e9bcf430c49ad4b684ba6aa90 Mon Sep 17 00:00:00 2001 From: Eike Stepper Date: Wed, 18 Apr 2012 18:14:02 +0200 Subject: [375939] Payload size of 21331 cause Buffer error when CDO UI is used https://bugs.eclipse.org/bugs/show_bug.cgi?id=375939 --- .../eclipse/net4j/internal/jvm/JVMConnector.java | 328 +++--- .../src/org/eclipse/net4j/tests/TransportTest.java | 1136 +++++++++++--------- .../org/eclipse/internal/net4j/buffer/Buffer.java | 870 +++++++-------- 3 files changed, 1209 insertions(+), 1125 deletions(-) (limited to 'plugins') 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 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 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 e = (IContainerEvent)event; - e.getDeltaElement().addListener(new IListener() - { - public void notifyEvent(IEvent event) - { - if (event instanceof IContainerEvent) - { - @SuppressWarnings("unchecked") - IContainerEvent e = (IContainerEvent)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 e = (IContainerEvent)event; - e.getDeltaElement().addListener(new IListener() - { - public void notifyEvent(IEvent event) - { - if (event instanceof IContainerEvent) - { - @SuppressWarnings("unchecked") - IContainerEvent e = (IContainerEvent)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 e = (IContainerEvent)event; - e.getDeltaElement().addListener(new IListener() - { - public void notifyEvent(IEvent event) - { - if (event instanceof IContainerEvent) - { - @SuppressWarnings("unchecked") - IContainerEvent e = (IContainerEvent)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 e = (IContainerEvent)event; - e.getDeltaElement().addListener(new IListener() - { - public void notifyEvent(IEvent event) - { - if (event instanceof IContainerEvent) - { - @SuppressWarnings("unchecked") - IContainerEvent e = (IContainerEvent)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 - { - 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 e = (IContainerEvent)event; + e.getDeltaElement().addListener(new IListener() + { + public void notifyEvent(IEvent event) + { + if (event instanceof IContainerEvent) + { + @SuppressWarnings("unchecked") + IContainerEvent e = (IContainerEvent)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 e = (IContainerEvent)event; + e.getDeltaElement().addListener(new IListener() + { + public void notifyEvent(IEvent event) + { + if (event instanceof IContainerEvent) + { + @SuppressWarnings("unchecked") + IContainerEvent e = (IContainerEvent)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 e = (IContainerEvent)event; + e.getDeltaElement().addListener(new IListener() + { + public void notifyEvent(IEvent event) + { + if (event instanceof IContainerEvent) + { + @SuppressWarnings("unchecked") + IContainerEvent e = (IContainerEvent)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 e = (IContainerEvent)event; + e.getDeltaElement().addListener(new IListener() + { + public void notifyEvent(IEvent event) + { + if (event instanceof IContainerEvent) + { + @SuppressWarnings("unchecked") + IContainerEvent e = (IContainerEvent)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 + { + 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 true if the buffer has been completely written, false 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 true if the buffer has been completely written, false 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(); + } + } +} -- cgit v1.2.1