[jboss-remoting-commits] JBoss Remoting SVN: r3452 - in remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp: id and 1 other directory.

Author: david.lloyd(a)jboss.com Date: 2008-02-18 21:10:12 -0500 (Mon, 18 Feb 2008) New Revision: 3452 Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppEndpointProtocolHandler.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandler.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandlerFactory.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoProvider.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppMessageType.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppProtocolBean.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppServerBean.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSession.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSessionProtocolHandler.java Modified: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/id/IdentifierManager.java Log: OK, this branch was a bad idea. Just commit the half-working stuff and I'll just merge back the one or two good ideas by hand... Modified: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java 2008-02-17 19:15:34 UTC (rev 3451) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -37,13 +37,6 @@ import org.jboss.cx.remoting.jrpp.id.JrppStreamIdentifier; import org.jboss.cx.remoting.jrpp.mina.IoBufferByteInput; import org.jboss.cx.remoting.jrpp.mina.IoBufferByteOutput; -import org.jboss.cx.remoting.spi.protocol.ContextIdentifier; -import org.jboss.cx.remoting.spi.protocol.ProtocolContext; -import org.jboss.cx.remoting.spi.protocol.ProtocolHandler; -import org.jboss.cx.remoting.spi.protocol.ProtocolServerContext; -import org.jboss.cx.remoting.spi.protocol.RequestIdentifier; -import org.jboss.cx.remoting.spi.protocol.ServiceIdentifier; -import org.jboss.cx.remoting.spi.protocol.StreamIdentifier; import javax.security.auth.callback.CallbackHandler; import javax.security.sasl.Sasl; @@ -66,8 +59,6 @@ private static final String SASL_SERVER_FILTER_NAME = "SASL server filter"; private IoSession ioSession; - private final ProtocolHandler protocolHandler; - private final ProtocolContext protocolContext; private final SingleSessionIoHandler ioHandler; private final IdentifierManager identifierManager; @@ -194,34 +185,10 @@ return ioHandler; } - public ProtocolHandler getProtocolHandler() { - return protocolHandler; - } - - public ProtocolContext getProtocolContext() { - return protocolContext; - } - private void write(ObjectOutput output, MessageType messageType) throws IOException { output.writeByte(messageType.ordinal()); } - private void write(ObjectOutput output, ServiceIdentifier serviceIdentifier) throws IOException { - output.writeShort(((JrppServiceIdentifier)serviceIdentifier).getId()); - } - - private void write(ObjectOutput output, ContextIdentifier contextIdentifier) throws IOException { - output.writeShort(((JrppContextIdentifier)contextIdentifier).getId()); - } - - private void write(ObjectOutput output, StreamIdentifier streamIdentifier) throws IOException { - output.writeShort(((JrppStreamIdentifier)streamIdentifier).getId()); - } - - private void write(ObjectOutput output, RequestIdentifier requestIdentifier) throws IOException { - output.writeShort(((JrppRequestIdentifier)requestIdentifier).getId()); - } - public void sendResponse(byte[] rawMsgData) throws IOException { final IoBuffer buffer = newBuffer(rawMsgData.length + 100, false); final MessageOutput output = protocolContext.getMessageOutput(new IoBufferByteOutput(buffer, ioSession)); Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppEndpointProtocolHandler.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppEndpointProtocolHandler.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppEndpointProtocolHandler.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,32 @@ +package org.jboss.cx.remoting.jrpp; + +import org.jboss.cx.remoting.spi.protocol.EndpointProtocolHandler; +import org.jboss.cx.remoting.spi.protocol.SessionProtocolHandler; +import org.jboss.cx.remoting.spi.protocol.SessionProtocolContext; +import org.jboss.cx.remoting.spi.protocol.EndpointProtocolContext; +import java.net.URI; +import java.io.IOException; + +/** + * + */ +public final class JrppEndpointProtocolHandler implements EndpointProtocolHandler { + private EndpointProtocolContext endpointProtocolContext; + + public JrppEndpointProtocolHandler() { + } + + public SessionProtocolHandler establishSession(SessionProtocolContext context, URI remoteURI) throws IOException { + final JrppSessionProtocolHandler sessionProtocolHandler = new JrppSessionProtocolHandler(context); + sessionProtocolHandler.connect(remoteURI); + return sessionProtocolHandler; + } + + public EndpointProtocolContext getEndpointProtocolContext() { + return endpointProtocolContext; + } + + public void setEndpointProtocolContext(final EndpointProtocolContext endpointProtocolContext) { + this.endpointProtocolContext = endpointProtocolContext; + } +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandler.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandler.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandler.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,14 @@ +package org.jboss.cx.remoting.jrpp; + +import org.jboss.cx.remoting.core.util.MessageInput; +import org.jboss.cx.remoting.core.util.MessageOutput; +import java.io.IOException; + +/** + * + */ +public interface JrppIoHandler { + void handleMessage(MessageInput message) throws IOException; + + MessageOutput getSaslMessageOutput(); +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandlerFactory.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandlerFactory.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoHandlerFactory.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,25 @@ +package org.jboss.cx.remoting.jrpp; + +import org.apache.mina.handler.multiton.SingleSessionIoHandlerFactory; +import org.apache.mina.handler.multiton.SingleSessionIoHandler; +import org.apache.mina.common.IoSession; +import org.jboss.cx.remoting.spi.protocol.EndpointProtocolContext; + +/** + * + */ +public final class JrppIoHandlerFactory implements SingleSessionIoHandlerFactory { + private final EndpointProtocolContext endpointProtocolContext; + + public JrppIoHandlerFactory(final EndpointProtocolContext endpointProtocolContext) { + this.endpointProtocolContext = endpointProtocolContext; + } + + public SingleSessionIoHandler getHandler(IoSession ioSession) throws Exception { + final JrppSession jrppSession = new JrppSession(); + jrppSession.setIoProvider(ioSession); + jrppSession.establishServer(); + endpointProtocolContext.esablishSession(jrppSession.getSessionProtocolHandler()); + return jrppSession.getSingleSessionIoHandler(); + } +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoProvider.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoProvider.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppIoProvider.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,24 @@ +package org.jboss.cx.remoting.jrpp; + +import org.jboss.cx.remoting.core.util.MessageOutput; +import org.jboss.cx.remoting.core.util.MessageInput; +import java.io.IOException; + +/** + * + */ +public interface JrppIoProvider { + MessageOutput createNewMessage(int estimatedSize) throws IOException; + + boolean receiveSaslChallenge(MessageInput data) throws IOException; + + boolean receiveSaslResponse(MessageInput data) throws IOException; + + boolean sendSaslInitialChallenge() throws IOException; + + boolean sendSaslInitialResponse() throws IOException; + + void close(); + + void startEncryption() throws IOException; +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppMessageType.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppMessageType.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppMessageType.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,25 @@ +package org.jboss.cx.remoting.jrpp; + +/** + * + */ +public enum JrppMessageType { + VERSION, /* 0 */ + SASL_CHALLENGE, + SASL_RESPONSE, + AUTH_SUCCESS, + AUTH_FAILED, + OPEN_CONTEXT, /* 5 */ + CANCEL_ACK, + CANCEL_REQ, + CLOSE_CONTEXT, + CLOSE_SERVICE, + CLOSE_STREAM, /* 10 */ + EXCEPTION, + REPLY, + REQUEST, + SERVICE_ACTIVATE, + SERVICE_REQUEST, /* 15 */ + SERVICE_TERMINATE, + STREAM_DATA, +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppProtocolBean.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppProtocolBean.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppProtocolBean.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,94 @@ +package org.jboss.cx.remoting.jrpp; + +import org.jboss.cx.remoting.Endpoint; +import org.jboss.cx.remoting.RemotingException; +import org.jboss.cx.remoting.spi.protocol.ProtocolRegistration; +import org.jboss.cx.remoting.spi.protocol.ProtocolRegistrationSpec; +import org.jboss.cx.remoting.spi.protocol.EndpointProtocolContext; + +/** + * + */ +public final class JrppProtocolBean { + /** + * The endpoint. + * + * @protectedby {@code this} + */ + private Endpoint endpoint; + + /** + * The protocol registration. + * + * @protectedby {@code this} + */ + private ProtocolRegistration registration; + + /** + * The endpoint protocol context. + * + * @protectedby {@code this} + */ + private EndpointProtocolContext endpointProtocolContext; + + public Endpoint getEndpoint() { + synchronized(this) { + return endpoint; + } + } + + public void setEndpoint(final Endpoint endpoint) { + synchronized(this) { + this.endpoint = endpoint; + } + } + + public ProtocolRegistration getRegistration() { + return registration; + } + + public EndpointProtocolContext getEndpointProtocolContext() { + return endpointProtocolContext; + } + + // Lifecycle methods + + public void create() throws RemotingException { + final JrppEndpointProtocolHandler endpointProtocolHandler = new JrppEndpointProtocolHandler(); + final ProtocolRegistrationSpec spec = ProtocolRegistrationSpec.DEFAULT.setScheme("jrpp").setEndpointProtocolHandler(endpointProtocolHandler); + final ProtocolRegistration registration; + final EndpointProtocolContext endpointProtocolContext; + synchronized(this) { + registration = endpoint.registerProtocol(spec); + this.registration = registration; + } + endpointProtocolContext = registration.getEndpointProtocolContext(); + this.endpointProtocolContext = endpointProtocolContext; + endpointProtocolHandler.setEndpointProtocolContext(endpointProtocolContext); + } + + public void start() { + final ProtocolRegistration registration; + synchronized(this) { + registration = this.registration; + } + registration.start(); + } + + public void stop() { + final ProtocolRegistration registration; + synchronized(this) { + registration = this.registration; + } + registration.stop(); + } + + public void destroy() { + final ProtocolRegistration registration; + synchronized(this) { + registration = this.registration; + this.registration = null; + } + registration.unregister(); + } +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppServerBean.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppServerBean.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppServerBean.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,98 @@ +package org.jboss.cx.remoting.jrpp; + +import org.apache.mina.common.IoAcceptor; +import org.apache.mina.transport.socket.nio.NioSocketAcceptor; +import org.apache.mina.transport.socket.nio.NioProcessor; +import org.apache.mina.handler.multiton.SingleSessionIoHandlerDelegate; +import org.jboss.cx.remoting.jrpp.mina.FramingIoFilter; +import java.util.concurrent.ExecutorService; +import java.net.SocketAddress; +import java.io.IOException; + +/** + * + */ +public final class JrppServerBean { + /** + * @protectedby {@code this} + */ + private JrppProtocolBean protocolBean; + /** + * @protectedby {@code this} + */ + private ExecutorService threadPool; + /** + * @protectedby {@code this} + */ + private IoAcceptor acceptor; + /** + * @protectedby {@code this} + */ + private SocketAddress address; + + public JrppProtocolBean getProtocolBean() { + synchronized(this) { + return protocolBean; + } + } + + public void setProtocolBean(final JrppProtocolBean protocolBean) { + synchronized(this) { + this.protocolBean = protocolBean; + } + } + + public ExecutorService getThreadPool() { + synchronized(this) { + return threadPool; + } + } + + public void setThreadPool(final ExecutorService threadPool) { + synchronized(this) { + this.threadPool = threadPool; + } + } + + public SocketAddress getAddress() { + synchronized(this) { + return address; + } + } + + public void setAddress(final SocketAddress address) { + synchronized(this) { + this.address = address; + } + } + + // Lifecycle methods + + public void create() { + synchronized(this) { + final NioProcessor nioProcessor = new NioProcessor(threadPool); + final IoAcceptor acceptor = new NioSocketAcceptor(threadPool, nioProcessor); + acceptor.setDefaultLocalAddress(address); + acceptor.setHandler(new SingleSessionIoHandlerDelegate(new JrppIoHandlerFactory(protocolBean.getEndpointProtocolContext()))); + acceptor.getFilterChain().addLast("framing filter", new FramingIoFilter()); + this.acceptor = acceptor; + } + } + + public void start() throws IOException { + acceptor.bind(address); + } + + public void stop() { + acceptor.unbind(); + } + + public void destroy() { + final IoAcceptor acceptor; + synchronized(this) { + acceptor = this.acceptor; + this.acceptor = null; + } + acceptor.dispose(); + } +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSession.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSession.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSession.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,343 @@ +package org.jboss.cx.remoting.jrpp; + +import org.jboss.cx.remoting.spi.protocol.SessionProtocolHandler; +import org.jboss.cx.remoting.spi.protocol.ServiceClientProtocolHandler; +import org.jboss.cx.remoting.spi.protocol.ServiceClientProtocolContext; +import org.jboss.cx.remoting.spi.protocol.ServiceProtocolContext; +import org.jboss.cx.remoting.spi.protocol.ServerContextProtocolContext; +import org.jboss.cx.remoting.spi.protocol.ClientRequestProtocolContext; +import org.jboss.cx.remoting.core.util.MessageInput; +import org.jboss.cx.remoting.core.util.MessageOutput; +import org.jboss.cx.remoting.core.util.AtomicStateMachine; +import org.jboss.cx.remoting.core.util.CollectionUtil; +import org.jboss.cx.remoting.ServiceLocator; +import org.jboss.cx.remoting.Endpoint; +import org.jboss.cx.remoting.RemoteExecutionException; +import org.jboss.cx.remoting.log.Logger; +import java.io.IOException; +import static java.lang.Math.min; +import java.util.Set; + +import javax.security.sasl.SaslException; + +/** + * + */ +public final class JrppSession { + /** + * The protocol version used by this version of Remoting. Value is transmitted as an unsigned short. + */ + private static final int PROTOCOL_VERSION = 0x0000; + + private static final Logger log = Logger.getLogger(JrppSession.class); + + private final Endpoint localEndpoint; + + private JrppIoProvider ioProvider; + private SessionProtocolHandler sessionProtocolHandler; + private JrppIoHandler ioHandler; + + private int authRetries = 3; + + private int protocolVersion; + private String remoteName; + + private enum State { + /** Not yet connected */ + INITIAL, + /** Client side, waiting to receive protocol version info */ + AWAITING_SERVER_VERSION, + /** Server side, waiting to receive protocol version info */ + AWAITING_CLIENT_VERSION, + /** Client side, auth phase */ + AWAITING_SERVER_CHALLENGE, + /** Server side, auth phase */ + AWAITING_CLIENT_RESPONSE, + /** Connection is up */ + UP, + /** Session is shutting down or closed */ + CLOSED, + } + + private final AtomicStateMachine<State> state = AtomicStateMachine.start(State.INITIAL); + + public JrppSession(final Endpoint localEndpoint) { + this.localEndpoint = localEndpoint; + } + + // accessors + + public void setIoProvider(final JrppIoProvider ioProvider) { + this.ioProvider = ioProvider; + } + + public JrppIoProvider getIoProvider() { + return ioProvider; + } + + // message util methods - write + + private void write(MessageOutput messageOutput, JrppMessageType type) throws IOException { + messageOutput.writeByte(type.ordinal()); + } + + // message util methods - read + + private JrppMessageType readMessageType(MessageInput input) throws IOException { + final int typeId = input.readByte() & 0xff; + final JrppMessageType[] types = JrppMessageType.values(); + if (types.length < typeId) { + throw new IOException("Received an invalid message type: " + typeId); + } + return types[typeId]; + } + + // init methods + + private void writeVersionMessage(State newState) throws IOException { + state.requireTransitionExclusive(State.INITIAL, newState); + boolean ok = false; + try { + final MessageOutput message = ioProvider.createNewMessage(40); + write(message, JrppMessageType.VERSION); + message.writeShort(PROTOCOL_VERSION); + final String name = localEndpoint.getName(); + message.writeUTF(name != null ? name : ""); + message.commit(); + ok = true; + } finally { + try { + if (! ok) close(); + } finally { + state.releaseExclusive(); + } + } + } + + public void establishServer() throws IOException { + writeVersionMessage(State.AWAITING_CLIENT_VERSION); + } + + public void establishClient() throws IOException { + writeVersionMessage(State.AWAITING_SERVER_VERSION); + } + + private void close() throws IOException { + if (state.transition(State.CLOSED)) { + ioProvider.close(); + } + } + + public SessionProtocolHandler getSessionProtocolHandler() { + return sessionProtocolHandler; + } + + public JrppIoHandler getIoHandler() { + return ioHandler; + } + + public final class Handler implements JrppIoHandler { + public void handleMessage(MessageInput message) throws IOException { + final JrppMessageType type = readMessageType(message); + final boolean trace = log.isTrace(); + final State current = state.getState(); + OUT: switch (current) { + case AWAITING_CLIENT_VERSION: { + switch (type) { + case VERSION: { + protocolVersion = min(message.readUnsignedShort(), PROTOCOL_VERSION); + if (trace) { + log.trace("Server negotiated protocol version " + protocolVersion); + } + final String name = message.readUTF(); + remoteName = name.length() > 0 ? name : null; + state.requireTransition(current, State.AWAITING_CLIENT_RESPONSE); + return; + } + default: break OUT; + } + } + case AWAITING_SERVER_VERSION: { + switch (type) { + case VERSION: { + protocolVersion = min(message.readUnsignedShort(), PROTOCOL_VERSION); + if (trace) { + log.trace("Client negotiated protocol version " + protocolVersion); + } + final String name = message.readUTF(); + remoteName = name.length() > 0 ? name : null; + state.requireTransition(current, State.AWAITING_SERVER_CHALLENGE); + ioProvider.sendSaslInitialResponse(); + return; + } + default: break OUT; + } + } + case AWAITING_CLIENT_RESPONSE: { + switch (type) { + case SASL_RESPONSE: { + if (trace) { + log.trace("Recevied SASL response from client"); + } + try { + if (ioProvider.receiveSaslResponse(message)) { + final MessageOutput output = ioProvider.createNewMessage(60); + write(output, JrppMessageType.AUTH_SUCCESS); + output.commit(); + ioProvider.startEncryption(); + } + } catch (SaslException ex) { + final MessageOutput output = ioProvider.createNewMessage(60); + write(output, JrppMessageType.AUTH_FAILED); + output.commit(); + log.info("Client authentication failed (" + ex.getMessage() + ")"); + } + return; + } + default: break OUT; + } + } + case AWAITING_SERVER_CHALLENGE: { + switch (type) { + case SASL_CHALLENGE: { + ioProvider.receiveSaslChallenge(message); + return; + } + case AUTH_SUCCESS: { + ioProvider.startEncryption(); + state.requireTransition(current, State.UP); + return; + } + case AUTH_FAILED: { + log.debug("JRPP client authentication failed"); + if (--authRetries > 0) { + ioProvider.sendSaslInitialResponse(); + } else { + close(); + } + return; + } + default: break OUT; + } + } + case UP: { + switch (type) { + case OPEN_CONTEXT: { + final int serviceId = message.readInt(); + ServiceProtocolContext serviceProtocolContext = null; // todo look it up + final int contextId = message.readInt(); + final ServerContextProtocolContext serverContextProtocolContext = serviceProtocolContext.openContext(); + // todo put the new context in the map + return; + } + case CANCEL_ACK: { + final int contextId = message.readInt(); + final int requestId = message.readInt(); + ClientRequestProtocolContext clientRequestProtocolContext = null; // todo look it up + clientRequestProtocolContext.receiveCancelled(); + // todo drop the request + return; + } + case CANCEL_REQ: { + final ContextIdentifier contextIdentifier = readCtxtId(message); + final RequestIdentifier requestIdentifier = readReqId(message); + final boolean mayInterrupt = message.readBoolean(); + protocolContext.receiveCancelRequest(contextIdentifier, requestIdentifier, mayInterrupt); + return; + } + case CLOSE_CONTEXT: { + final ContextIdentifier contextIdentifier = readCtxtId(message); + protocolContext.closeContext(contextIdentifier); + return; + } + case CLOSE_SERVICE: { + final ServiceIdentifier serviceIdentifier = readSvcId(message); + protocolContext.closeService(serviceIdentifier); + return; + } + case CLOSE_STREAM: { + final StreamIdentifier streamIdentifier = readStrId(message); + protocolContext.closeStream(streamIdentifier); + return; + } + case EXCEPTION: { + final ContextIdentifier contextIdentifier = readCtxtId(message); + final RequestIdentifier requestIdentifier = readReqId(message); + final RemoteExecutionException exception = (RemoteExecutionException) message.readObject(); + protocolContext.receiveException(contextIdentifier, requestIdentifier, exception); + return; + } + case REPLY: { + final ContextIdentifier contextIdentifier = readCtxtId(message); + final RequestIdentifier requestIdentifier = readReqId(message); + final Object reply = message.readObject(); + protocolContext.receiveReply(contextIdentifier, requestIdentifier, reply); + return; + } + case REQUEST: { + final ContextIdentifier contextIdentifier = readCtxtId(message); + final RequestIdentifier requestIdentifier = readReqId(message); + final Object request = message.readObject(); + if (trace) { + log.trace("Received request - body is %s", request); + } + protocolContext.receiveRequest(contextIdentifier, requestIdentifier, request); + return; + } + case SERVICE_ACTIVATE: { + final ServiceIdentifier serviceIdentifier = readSvcId(message); + protocolContext.receiveServiceActivate(serviceIdentifier); + return; + } + case SERVICE_REQUEST: { + final ServiceIdentifier serviceIdentifier = readSvcId(message); + final Class<?> requestType = (Class<?>) message.readObject(); + final Class<?> replyType = (Class<?>) message.readObject(); + final String serviceType = message.readUTF(); + final String serviceGroupName = message.readUTF(); + final Set<String> interceptors = CollectionUtil.hashSet(); + int c = message.readInt(); + for (int i = 0; i < c; i ++) { + interceptors.add(message.readUTF()); + } + final ServiceLocator<?, ?> locator = ServiceLocator.DEFAULT + .setRequestType(requestType) + .setReplyType(replyType) + .setServiceType(serviceType) + .setServiceGroupName(serviceGroupName) + .setAvailableInterceptors(interceptors); + protocolContext.receiveServiceRequest(serviceIdentifier, locator); + return; + } + case SERVICE_TERMINATE: { + final ServiceIdentifier serviceIdentifier = readSvcId(message); + protocolContext.receiveServiceTerminate(serviceIdentifier); + return; + } + case STREAM_DATA: { + final StreamIdentifier streamIdentifier = readStrId(message); + protocolContext.receiveStreamData(streamIdentifier, message); + return; + } + default: break OUT; + } + } + } + throw new IllegalStateException("Got message " + type + " during " + currentState); + } + + public MessageOutput getSaslMessageOutput() { + return null; + } + } + + public final class WritingHandler implements SessionProtocolHandler { + public ServiceClientProtocolHandler sendServiceRequest(ServiceLocator<?, ?> locator, ServiceClientProtocolContext context) throws IOException { + return null; + } + + public String getRemoteName() { + return null; + } + } +} Added: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSessionProtocolHandler.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSessionProtocolHandler.java (rev 0) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppSessionProtocolHandler.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -0,0 +1,30 @@ +package org.jboss.cx.remoting.jrpp; + +import org.jboss.cx.remoting.spi.protocol.SessionProtocolHandler; +import org.jboss.cx.remoting.spi.protocol.SessionProtocolContext; +import org.jboss.cx.remoting.spi.protocol.ServiceClientProtocolHandler; +import org.jboss.cx.remoting.spi.protocol.ServiceClientProtocolContext; +import org.jboss.cx.remoting.ServiceLocator; +import java.net.URI; +import java.io.IOException; + +/** + * + */ +public final class JrppSessionProtocolHandler implements SessionProtocolHandler { + public JrppSessionProtocolHandler(final SessionProtocolContext context) { + + } + + public ServiceClientProtocolHandler sendServiceRequest(ServiceLocator<?, ?> locator, ServiceClientProtocolContext context) throws IOException { + return null; + } + + public String getRemoteName() { + return null; + } + + public void connect(final URI remoteURI) throws IOException { + + } +} Modified: remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/id/IdentifierManager.java =================================================================== --- remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/id/IdentifierManager.java 2008-02-17 19:15:34 UTC (rev 3451) +++ remoting3/branches/protocol-context-rework/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/id/IdentifierManager.java 2008-02-19 02:10:12 UTC (rev 3452) @@ -6,8 +6,10 @@ * */ public final class IdentifierManager { - private final BitSet bitSet = new BitSet(64); + private static final int INITIAL_BIT_COUNT = 64; + private final BitSet bitSet = new BitSet(INITIAL_BIT_COUNT); + public synchronized short getIdentifier() { final int id = bitSet.nextClearBit(1); if (id > 0xffff) { @@ -20,8 +22,4 @@ public synchronized void freeIdentifier(short id) { bitSet.clear(id & 0xffff); } - - public void getIdentifier(final short id) { - - } }