+ * Copyright Aug 14, 2009 + *
+ */ +public class BisocketControlConnectionReplacementTestCase extends TestCase +{ + private static Logger log =3D Logger.getLogger(BisocketControlConnectio= nReplacementTestCase.class); + = + protected static boolean firstTime =3D true; + protected static int secondaryServerSocketPort; + protected static int numberOfCallbacks =3D 10; + protected static Object lock =3D new Object(); + protected static TestCallbackHandler testCallbackHandler; + = + 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 =3D false; + Logger.getLogger("org.jboss.remoting").setLevel(XLevel.DEBUG); + Logger.getLogger("org.jboss.test.remoting").setLevel(Level.INFO); + String pattern =3D "[%d{ABSOLUTE}] [%t] %5p (%F:%L) - %m%n"; + PatternLayout layout =3D new PatternLayout(pattern); + ConsoleAppender consoleAppender =3D new ConsoleAppender(layout); + Logger.getRootLogger().addAppender(consoleAppender); = + } + } + + = + public void tearDown() + { + } + = + = + public void testCreateSocketWithReplacedControlConnection() throws Thro= wable + { + log.info("entering " + getName()); + = + // Start server. + setupServer(); + = + // Create client. + InvokerLocator clientLocator =3D new InvokerLocator(locatorURI); + HashMap clientConfig =3D new HashMap(); + clientConfig.put(InvokerLocator.FORCE_REMOTE, "true"); + addExtraClientConfig(clientConfig); + Client client =3D new Client(clientLocator, clientConfig); + client.connect(); + log.info("client is connected"); + = + // Test connections. + assertEquals("abc", client.invoke("abc")); + log.info("connection is good"); + = + // Add callback handler. + testCallbackHandler =3D new TestCallbackHandler(); + HashMap metadata =3D new HashMap(); + metadata.put(Bisocket.IS_CALLBACK_SERVER, "true"); + client.addListener(testCallbackHandler, metadata); + = + synchronized (lock) + { + lock.wait(120000); + } + = + assertEquals(numberOfCallbacks, testCallbackHandler.counter); + assertEquals(numberOfCallbacks - 1, testCallbackHandler.max); + = + client.removeListener(testCallbackHandler); + client.disconnect(); + shutdownServer(); + log.info(getName() + " PASSES"); + } + = + = + protected String getTransport() + { + return "bisocket"; + } + = + = + protected String getServerSocketName() + { + return TestServerSocketFactory.class.getName(); + } + = + = + protected void addExtraClientConfig(Map config) {} + protected void addExtraServerConfig(Map config) {} + = + + protected void setupServer() throws Exception + { + host =3D InetAddress.getLocalHost().getHostAddress(); + port =3D PortUtil.findFreePort(host); + locatorURI =3D getTransport() + "://" + host + ":" + port; + locatorURI +=3D "/?" + Bisocket.PING_FREQUENCY + "=3D2000"; + locatorURI +=3D "&" + DefaultCallbackErrorHandler.CALLBACK_ERRORS_AL= LOWED + "=3D100"; + String metadata =3D System.getProperty("remoting.metadata"); + if (metadata !=3D null) + { + locatorURI +=3D "&" + metadata; + } + serverLocator =3D new InvokerLocator(locatorURI); + log.info("Starting remoting server with locator uri of: " + locatorU= RI); + HashMap config =3D new HashMap(); + config.put(InvokerLocator.FORCE_REMOTE, "true"); + secondaryServerSocketPort =3D PortUtil.findFreePort(host); + config.put(Bisocket.SECONDARY_BIND_PORT, Integer.toString(secondaryS= erverSocketPort)); + config.put(ServerInvoker.SERVER_SOCKET_FACTORY, getServerSocketName(= )); + addExtraServerConfig(config); + connector =3D new Connector(serverLocator, config); + connector.create(); + invocationHandler =3D new TestInvocationHandler(); + connector.addInvocationHandler("test", invocationHandler); + connector.start(); + } + = + = + protected void shutdownServer() throws Exception + { + if (connector !=3D null) + connector.stop(); + } + = + = + static class TestInvocationHandler implements ServerInvocationHandler + { + int counter; + = + public void addListener(final InvokerCallbackHandler callbackHandler) + { + if (counter++ > 0) + return; + = + new Thread() + { + public void run() + { + for (int i =3D 0; i < 10 * numberOfCallbacks; i++) + { + try + { + if (testCallbackHandler.counter >=3D numberOfCallback= s) + { + return; + } + try + { + Thread.sleep(1000); + } + catch (InterruptedException e) + { + log.error("Unexpected interrupt", e); + } + log.info("sending callback: " + i); + callbackHandler.handleCallback(new Callback(Integer.t= oString(i))); + log.info("sent callback: " + i); + } + catch (HandleCallbackException e) + { + log.error("Callback error", e); + } + } = + } + }.start(); + } + public Object invoke(final InvocationRequest invocation) throws Thro= wable + { + return invocation.getParameter(); + } + public void removeListener(InvokerCallbackHandler callbackHandler) {} + public void setMBeanServer(MBeanServer server) {} + public void setInvoker(ServerInvoker invoker) {} + } + = + = + static class TestCallbackHandler implements InvokerCallbackHandler + { + int counter; + int max; + = + public void handleCallback(Callback callback) throws HandleCallbackE= xception + { + log.info("received callback: " + counter++); + max =3D Math.max(Integer.valueOf((String) callback.getParameter()= ).intValue(), max); + log.info("max: " + max); + if (counter >=3D numberOfCallbacks) + { + synchronized (lock) + { + lock.notifyAll(); + } + } + } = + } + = + static public class TestServerSocketFactory extends ServerSocketFactory + { + int timeout; + int initialWrites; + = + public TestServerSocketFactory() + { + this.timeout =3D 5000; + this.initialWrites =3D 2; + } = + public TestServerSocketFactory(int timeout, int initialWrites) + { + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public ServerSocket createServerSocket() throws IOException + { + ServerSocket ss =3D new TestServerSocket(timeout, initialWrites); + log.info("returning: " + ss); + return ss; + } + public ServerSocket createServerSocket(int port) throws IOException + { + ServerSocket ss =3D null; + if (port !=3D secondaryServerSocketPort) + { + ss =3D ServerSocketFactory.getDefault().createServerSocket(por= t); + } + else + { + ss =3D new TestServerSocket(port, timeout, initialWrites); + } + log.info("returning: " + ss); + return ss; + } + + public ServerSocket createServerSocket(int port, int backlog) throws= IOException + { + ServerSocket ss =3D null; + if (port !=3D secondaryServerSocketPort) + { + ss =3D ServerSocketFactory.getDefault().createServerSocket(por= t, backlog); + } + else + { + ss =3D new TestServerSocket(port, backlog, timeout, initialWri= tes); + } + log.info("returning: " + ss); + return ss; + } + + public ServerSocket createServerSocket(int port, int backlog, InetAd= dress ifAddress) throws IOException + { + ServerSocket ss =3D null; + if (port !=3D secondaryServerSocketPort) + { + ss =3D ServerSocketFactory.getDefault().createServerSocket(por= t, backlog, ifAddress); + } + else + { + ss =3D new TestServerSocket(port, backlog, ifAddress, timeout,= initialWrites); + } + log.info("returning: " + ss); + return ss; + } + } + = + = + static class TestServerSocket extends ServerSocket + { + int timeout; + int initialWrites; + + public TestServerSocket(int timeout, int initialWrites) throws IOExc= eption + { + super(); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestServerSocket(int port, int timeout, int initialWrites) th= rows IOException + { + super(port); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestServerSocket(int port, int backlog, int timeout, int init= ialWrites) throws IOException + { + super(port, backlog); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestServerSocket(int port, int backlog, InetAddress bindAddr,= int timeout, int initialWrites) throws IOException + { + super(port, backlog, bindAddr); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public Socket accept() throws IOException + { + Socket s =3D new TestSocket(timeout, initialWrites); + implAccept(s); + return s; + } + public String toString() + { + return "TestServerSocket[" + getLocalPort() + "]"; + } + } + = + = + static class TestSocket extends Socket + { + int timeout; + int initialWrites; + = + public TestSocket(int timeout, int initialWrites) + { + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestSocket(String host, int port, int timeout, int initialWri= tes) throws UnknownHostException, IOException + { + super(host, port); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestSocket(InetAddress address, int port, int timeout, int in= itialWrites) throws IOException + { + super(address, port); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestSocket(String host, int port, InetAddress localAddr, int = localPort, int timeout, int initialWrites) throws IOException + { + super(host, port, localAddr, localPort); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public TestSocket(InetAddress address, int port, InetAddress localAd= dr, int localPort, int timeout, int initialWrites) throws IOException + { + super(address, port, localAddr, localPort); + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public OutputStream getOutputStream() throws IOException + { + return new TestOutputStream(super.getOutputStream(), timeout, ini= tialWrites); + } + public String toString() + { + return "TestSocket[" + getLocalPort() + "->" + getPort() + "]"; + } + } + = + public static class TestOutputStream extends OutputStream + { + OutputStream os; + int timeout; + boolean closed; + int initialWrites; + boolean doCounterTest =3D true; + int counter; + = + public TestOutputStream(OutputStream os, int timeout, int initialWri= tes) + { + this.os =3D os; + this.timeout =3D timeout; + this.initialWrites =3D initialWrites; + } + public void close()throws IOException + { + closed =3D true; + super.close(); + log.info(this + " closed"); + } + public void write(int b) throws IOException + { + if (closed) + { + log.info("TestOutputStream closed, cannot write"); + throw new SocketException("closed"); + } + if (doCounterTest && ++counter > initialWrites) + { + close(); + throw new SocketException("closed"); + } + os.write(b); + } + public void write(byte b[], int off, int len) throws IOException + { + if (closed) + { + log.info("TestOutputStream closed, cannot write"); + throw new SocketException("closed"); + } + log.info("TestOutputStream: counter =3D " + counter + ", initialW= rites =3D " + initialWrites); + if (++counter > initialWrites) + { + close(); + throw new SocketException("closed"); + } + try + { + log.info(this + " writing"); + doCounterTest =3D false; + os.write(b, off, len); + doCounterTest =3D true; + log.info(this + " back from writing"); + } + catch (IOException e) + { + log.info("exception: ", e); + throw e; + } + } + } +} \ No newline at end of file --===============5781342472527020376==--