Author: ron.sigal(a)jboss.com Date: 2013-01-29 01:56:09 -0500 (Tue, 29 Jan 2013) New Revision: 6478 Added: remoting2/branches/2.x/src/tests/org/jboss/test/remoting/transport/socket/idempotent/ remoting2/branches/2.x/src/tests/org/jboss/test/remoting/transport/socket/idempotent/IdempotentTestCase.java Modified: remoting2/branches/2.x/src/main/org/jboss/remoting/transport/socket/MicroSocketClientInvoker.java Log: JBREM-1316: Added idempotent parameter to MicroSocketClientInvoker. Modified: remoting2/branches/2.x/src/main/org/jboss/remoting/transport/socket/MicroSocketClientInvoker.java =================================================================== --- remoting2/branches/2.x/src/main/org/jboss/remoting/transport/socket/MicroSocketClientInvoker.java 2013-01-10 07:00:56 UTC (rev 6477) +++ remoting2/branches/2.x/src/main/org/jboss/remoting/transport/socket/MicroSocketClientInvoker.java 2013-01-29 06:56:09 UTC (rev 6478) @@ -264,6 +264,11 @@ protected boolean generalizeSocketException; protected int writeTimeout = -1; + + /** + * Determines if transport() should attempt retries after versionedWrite() has succeeded. + */ + protected boolean idempotent; // Constructors --------------------------------------------------------------------------------- @@ -286,7 +291,8 @@ pool = null; maxPoolSize = MAX_POOL_SIZE; onewayConnectionTimeout = ONEWAY_CONNECTION_TIMEOUT_DEFAULT; - + idempotent = true; + try { setup(); @@ -421,6 +427,16 @@ this.generalizeSocketException = generalizeSocketException; } + public boolean isIdempotent() + { + return idempotent; + } + + public void setIdempotent(boolean idempotent) + { + this.idempotent = idempotent; + } + public synchronized void disconnect() { log.debug(this + " disconnecting ..."); @@ -795,8 +811,9 @@ int retryCount = 0; Exception sockEx = null; + boolean versionedWriteComplete = false; - for (; retryCount < numberOfCallRetries; retryCount++) + for (; retryCount < numberOfCallRetries && (idempotent || !versionedWriteComplete); retryCount++) { if (trace) log.trace(this + " retryCount: " + retryCount); if (0 < tempTimeout) @@ -866,7 +883,8 @@ //TODO: -TME so this is messed up as now ties remoting versioning to using a marshaller type versionedWrite(outputStream, marshaller, invocation, version); - + versionedWriteComplete = true; + if (serverSideOneway) { if(trace) { log.trace(this + " sent oneway invocation, so not waiting for response, returning null"); } @@ -958,7 +976,7 @@ } // need to check if ran out of retries - if (retryCount >= numberOfCallRetries) + if (retryCount >= numberOfCallRetries || (response == null && !idempotent && versionedWriteComplete)) { handleException(sockEx, socketWrapper); } @@ -1289,9 +1307,7 @@ return null; } - // Private -------------------------------------------------------------------------------------- - - private Object versionedRead(InputStream inputStream, UnMarshaller unmarshaller, int version) + protected Object versionedRead(InputStream inputStream, UnMarshaller unmarshaller, int version) throws IOException, ClassNotFoundException { //TODO: -TME - is switch required? @@ -1315,7 +1331,7 @@ } } - private void versionedWrite(OutputStream outputStream, Marshaller marshaller, + protected void versionedWrite(OutputStream outputStream, Marshaller marshaller, Object invocation, int version) throws IOException { //TODO: -TME Should I worry about checking the version here? Only one way to do it at this point @@ -1343,7 +1359,7 @@ } //TODO: -TME Exact same method in ServerThread - private int readVersion(InputStream inputStream) throws IOException + protected int readVersion(InputStream inputStream) throws IOException { if (trace) { log.trace(this + " reading version from input stream"); } int version = inputStream.read(); @@ -1352,7 +1368,7 @@ } //TODO: -TME Exact same method in ServerThread - private void writeVersion(OutputStream outputStream, int version) throws IOException + protected void writeVersion(OutputStream outputStream, int version) throws IOException { if (trace) { log.trace(this + " writing version " + version + " on output stream"); } outputStream.write(version); Added: remoting2/branches/2.x/src/tests/org/jboss/test/remoting/transport/socket/idempotent/IdempotentTestCase.java =================================================================== --- remoting2/branches/2.x/src/tests/org/jboss/test/remoting/transport/socket/idempotent/IdempotentTestCase.java (rev 0) +++ remoting2/branches/2.x/src/tests/org/jboss/test/remoting/transport/socket/idempotent/IdempotentTestCase.java 2013-01-29 06:56:09 UTC (rev 6478) @@ -0,0 +1,680 @@ +/* +* JBoss, Home of Professional Open Source +* Copyright 2009, JBoss Inc., and individual contributors as indicated +* by the @authors tag. See the copyright.txt in the distribution for a +* full listing of individual contributors. +* +* This is free software; you can redistribute it and/or modify it +* under the terms of the GNU Lesser General Public License as +* published by the Free Software Foundation; either version 2.1 of +* the License, or (at your option) any later version. +* +* This software is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +* Lesser General Public License for more details. +* +* You should have received a copy of the GNU Lesser General Public +* License along with this software; if not, write to the Free +* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA +* 02110-1301 USA, or see the FSF site: http://www.fsf.org. +*/ +package org.jboss.test.remoting.transport.socket.idempotent; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.SocketException; +import java.util.HashMap; +import java.util.Map; + +import javax.management.MBeanServer; + +import junit.framework.TestCase; + +import org.apache.log4j.ConsoleAppender; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.apache.log4j.PatternLayout; +import org.jboss.remoting.Client; +import org.jboss.remoting.InvocationFailureException; +import org.jboss.remoting.InvocationRequest; +import org.jboss.remoting.InvokerLocator; +import org.jboss.remoting.InvokerRegistry; +import org.jboss.remoting.ServerInvocationHandler; +import org.jboss.remoting.ServerInvoker; +import org.jboss.remoting.callback.InvokerCallbackHandler; +import org.jboss.remoting.marshal.Marshaller; +import org.jboss.remoting.marshal.UnMarshaller; +import org.jboss.remoting.transport.ClientFactory; +import org.jboss.remoting.transport.ClientInvoker; +import org.jboss.remoting.transport.Connector; +import org.jboss.remoting.transport.PortUtil; +import org.jboss.remoting.transport.socket.SocketClientInvoker; +import org.jboss.remoting.transport.socket.SocketWrapper; +import org.jboss.remoting.transport.socket.TransportServerFactory; + + +/** + * JBREM-1316 + * + * @author <a href="mailto:ron.sigal@jboss.com">Ron Sigal</a> + * @version $Rev$ + * <p> + * Copyright Oct 24, 2012 + * </p> + */ +public class IdempotentTestCase extends TestCase +{ + private static final int NO_FAILURE = 0; + private static final int FAIL_DURING_CONNECT = 1; + private static final int FAIL_DURING_WRITE_VERSION = 2; + private static final int FAIL_DURING_VERSIONED_WRITE = 3; + private static final int FAIL_AFTER_VERSIONED_WRITE = 4; + + private static Logger log = Logger.getLogger(IdempotentTestCase.class); + + private static boolean firstTime = true; + private static int state; + private static boolean retry = false; + + protected String host; + protected int port; + protected String locatorURI; + protected InvokerLocator serverLocator; + protected Connector connector; + protected TestInvocationHandler invocationHandler; + + public void setUp() throws Exception + { + if (firstTime) + { + firstTime = false; + Logger.getLogger("org.jboss.remoting").setLevel(Level.INFO); + Logger.getLogger("org.jboss.test.remoting").setLevel(Level.INFO); + String pattern = "[%d{ABSOLUTE}] [%t] %5p (%F:%L) - %m%n"; + PatternLayout layout = new PatternLayout(pattern); + ConsoleAppender consoleAppender = new ConsoleAppender(layout); + Logger.getRootLogger().addAppender(consoleAppender); + } + InvokerRegistry.registerInvokerFactories(getTransport(), TestClientFactory.class, TransportServerFactory.class); + } + + + public void tearDown() + { + } + + + public void testDefaultNoFailures() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?blue=monkey", NO_FAILURE); + assertEquals("abc", client.invoke("abc")); + assertEquals(1, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testDefaultFailureDuringConnect() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?blue=monkey", FAIL_DURING_CONNECT); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testDefaultFailureDuringWriteVersion() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?blue=monkey", FAIL_DURING_WRITE_VERSION); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testDefaultFailureDuringVersionedWrite() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?blue=monkey", FAIL_DURING_VERSIONED_WRITE); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(2, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testDefaultFailureAfterVersionedWrite() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?blue=monkey", FAIL_AFTER_VERSIONED_WRITE); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(2, TestClientInvoker.versionedWriteCounter); + assertEquals(2, TestClientInvoker.readVersionCounter); + assertEquals(2, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentTrueNoFailures() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=true", NO_FAILURE); + assertEquals("abc", client.invoke("abc")); + assertEquals(1, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentTrueFailureDuringConnect() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=true", FAIL_DURING_CONNECT); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentTrueFailureDuringWriteVersion() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=true", FAIL_DURING_WRITE_VERSION); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentTrueFailureDuringVersionedWrite() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=true", FAIL_DURING_VERSIONED_WRITE); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(2, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentTrueFailureAfterVersionedWrite() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=true", FAIL_AFTER_VERSIONED_WRITE); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(2, TestClientInvoker.versionedWriteCounter); + assertEquals(2, TestClientInvoker.readVersionCounter); + assertEquals(2, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentFalseNoFailures() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=false", NO_FAILURE); + assertEquals("abc", client.invoke("abc")); + assertEquals(1, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentFalseFailureDuringConnect() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=false", FAIL_DURING_CONNECT); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentFalseFailureDuringWriteVersion() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=false", FAIL_DURING_WRITE_VERSION); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentFalseFailureDuringVersionedWrite() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=false", FAIL_DURING_VERSIONED_WRITE); + assertEquals("abc", client.invoke("abc")); + assertEquals(2, TestClientInvoker.connectionCounter); + assertEquals(2, TestClientInvoker.writeVersionCounter); + assertEquals(2, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + public void testIdempotentFalseFailureAfterVersionedWrite() throws Throwable + { + log.info("entering " + getName()); + Client client = setup("/?idempotent=false", FAIL_AFTER_VERSIONED_WRITE); + try + { + client.invoke("abc"); + fail("expected exception"); + } + catch (InvocationFailureException e) + { + // OK + } + catch (Exception e) + { + fail("expected InvocationFailureException"); + } + assertEquals(1, TestClientInvoker.connectionCounter); + assertEquals(1, TestClientInvoker.writeVersionCounter); + assertEquals(1, TestClientInvoker.versionedWriteCounter); + assertEquals(1, TestClientInvoker.readVersionCounter); + assertEquals(1, invocationHandler.counter); + retry = true; + assertEquals("xyz", client.invoke("xyz")); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + + + protected String getTransport() + { + return "socket"; + } + + + protected void addExtraClientConfig(Map config) {} + protected void addExtraServerConfig(Map config) {} + + + protected Client setup(String idempotentConfig, int nextState) throws Exception + { + // Reset counters. + TestClientInvoker.connectionCounter = 0; + TestClientInvoker.readVersionCounter = 0; + TestClientInvoker.writeVersionCounter = 0; + TestClientInvoker.versionedWriteCounter = 0; + + // Start server. + setupServer(idempotentConfig); + + // Create client. + InvokerLocator clientLocator = new InvokerLocator(locatorURI); + HashMap clientConfig = new HashMap(); + clientConfig.put(InvokerLocator.FORCE_REMOTE, "true"); + addExtraClientConfig(clientConfig); + Client client = new Client(clientLocator, clientConfig); + client.connect(); + log.info("client is connected"); + + // Set state. + state = nextState; + retry = false; + return client; + } + + + protected void setupServer(String idempotentConfig) throws Exception + { + host = InetAddress.getLocalHost().getHostAddress(); + port = PortUtil.findFreePort(host); + locatorURI = getTransport() + "://" + host + ":" + port + idempotentConfig; + String metadata = System.getProperty("remoting.metadata"); + if (metadata != null) + { + locatorURI += "&" + metadata; + } + serverLocator = new InvokerLocator(locatorURI); + log.info("Starting remoting server with locator uri of: " + locatorURI); + HashMap config = new HashMap(); + config.put(InvokerLocator.FORCE_REMOTE, "true"); + addExtraServerConfig(config); + connector = new Connector(serverLocator, config); + connector.create(); + invocationHandler = new TestInvocationHandler(); + connector.addInvocationHandler("test", invocationHandler); + connector.start(); + } + + + protected void shutdownServer() throws Exception + { + if (connector != null) + connector.stop(); + } + + + static class TestInvocationHandler implements ServerInvocationHandler + { + public int counter; + + public void addListener(InvokerCallbackHandler callbackHandler) {} + public Object invoke(final InvocationRequest invocation) throws Throwable + { + log.info(this + ".invoke() called"); + counter++; + return invocation.getParameter(); + } + public void removeListener(InvokerCallbackHandler callbackHandler) {} + public void setMBeanServer(MBeanServer server) {} + public void setInvoker(ServerInvoker invoker) {} + } + + +// static class TestServerThread extends ServerThread +// { +// static int counter; +// static Method readVersion; +// +// static +// { +// try +// { +// readVersion = ServerThread.class.getDeclaredMethod("readVersion", new Class[] {InputStream.class}); +// } +// catch (NoSuchMethodException e) +// { +// throw new RuntimeException("Couldn't find readVersion()"); +// } +// readVersion.setAccessible(true); +// } +// +// public TestServerThread(Socket socket, SocketServerInvoker invoker, LRUPool clientpool, +// LinkedList threadpool, int timeout, int writeTimeout, String serverSocketClassName) +// throws Exception +// { +// super(socket, invoker, clientpool, threadpool, timeout, writeTimeout, serverSocketClassName); +// } +// +// protected void processInvocation(SocketWrapper socketWrapper, InputStream inputStream, OutputStream outputStream) throws Exception +// { +// if (performVersioning) +// { +// version = readVersion(inputStream); +// if(version == -1) +// { +// throw new EOFException(); +// } +// } +// +// completeInvocation(socketWrapper, inputStream, outputStream, performVersioning, version); +// } +// +// private int readVersion(InputStream inputStream) throws Exception +// { +//// new Exception("failed: " + failed).printStackTrace(); +//// log.info("version = " + version + ", state = " + state); +// if (counter == 0 && state == FAIL_DURING_WRITE_VERSION) +// { +// counter++; +// throw new SocketException(); +// } +// else +// { +// Integer version = (Integer) readVersion.invoke(this, new Object[] {inputStream}); +// log.info("version: " + version.intValue()); +// if (version.intValue() != -1) +// { +// counter++; +// } +// return version.intValue(); +// } +// } +// +// protected Object versionedRead(InputStream inputStream, ServerInvoker invoker, +// ClassLoader classLoader, int version) throws IOException, ClassNotFoundException +// { +// counter++; +// if (counter == 2 && state == FAIL_DURING_VERSIONED_WRITE) +// { +// throw new SocketException(); +// } +// else if (counter == 2 && state == FAIL_AFTER_VERSIONED_WRITE) +// { +// super.versionedRead(inputStream, invoker, classLoader, version); +// throw new SocketException(); +// } +// else +// { +// return super.versionedRead(inputStream, invoker, classLoader, version); +// } +// } +// } +// +// +// static class TestServerInvoker extends SocketServerInvoker +// { +// public static int counter; +// +// public TestServerInvoker(InvokerLocator locator, Map configuration) +// { +// super(locator, configuration); +// } +// +// public TestServerInvoker(InvokerLocator locator) +// { +// super(locator); +// } +// +// protected void processInvocation(Socket socket) throws Exception +// { +// counter++; +// if (counter == 1 && state == FAIL_DURING_CONNECT) +// { +// log.info("closing socket"); +// socket.close(); +// } +// else +// { +// ServerThread worker = new TestServerThread(socket, this, clientpool, threadpool, getTimeout(), writeTimeout, serverSocketClass); +// worker.start(); +// } +// } +// } +// +// +// public static class TestServerFactory implements ServerFactory +// { +// public ServerInvoker createServerInvoker(InvokerLocator locator, Map config) throws IOException +// { +// log.info("TestServerFactory.createServerInvoker() called"); +// return new TestServerInvoker(locator, config); +// } +// public boolean supportsSSL() +// { +// return false; +// } +// } +// + + public static class TestClientInvoker extends SocketClientInvoker + { + public static int connectionCounter; + public static int writeVersionCounter; + public static int versionedWriteCounter; + public static int readVersionCounter; + + public TestClientInvoker(InvokerLocator locator, Map configuration) + { + super(locator, configuration); + } + + protected SocketWrapper getConnection(Marshaller marshaller, + UnMarshaller unmarshaller, + boolean tryPool, int timeAllowed) throws Exception + { + if (connectionCounter++ == 0 && state == FAIL_DURING_CONNECT) + { + throw new SocketException(); + } + SocketWrapper s = super.getConnection(marshaller, unmarshaller, tryPool, timeAllowed); + log.info(s.getSocket().getLocalPort() + ""); + log.info(s.getSocket().getPort() + ""); + return s; + } + + protected void writeVersion(OutputStream outputStream, int version) throws IOException + { + if (!retry && writeVersionCounter++ == 0 && state == FAIL_DURING_WRITE_VERSION) + { + throw new SocketException(); + } + super.writeVersion(outputStream, version); + log.info("client wrote version"); + } + + protected void versionedWrite(OutputStream outputStream, Marshaller marshaller, + Object invocation, int version) throws IOException + { + if (!retry && versionedWriteCounter++ == 0 && state == FAIL_DURING_VERSIONED_WRITE) + { + throw new SocketException(); + } + super.versionedWrite(outputStream, marshaller, invocation, version); + log.info("client wrote invocation"); + } + + protected int readVersion(InputStream inputStream) throws IOException + { + if (!retry && readVersionCounter++ == 0 && state == FAIL_AFTER_VERSIONED_WRITE) + { + try + { + // Give server a chance to read the invocation. + Thread.sleep(1000); + } + catch (InterruptedException e) + { + e.printStackTrace(); + } + throw new SocketException(); + } + return super.readVersion(inputStream); + } + } + + + public static class TestClientFactory implements ClientFactory + { + public ClientInvoker createClientInvoker(InvokerLocator locator, Map config) throws IOException + { + return new TestClientInvoker(locator, config); + } + public boolean supportsSSL() + { + return false; + } + } +} \ No newline at end of file