Transport Protocols

/**
 * STOMP protocol constants and command definitions
 */
public interface Stomp {
    /** STOMP commands */
    String CONNECT = "CONNECT";
    String CONNECTED = "CONNECTED";
    String SEND = "SEND";
    String SUBSCRIBE = "SUBSCRIBE";
    String UNSUBSCRIBE = "UNSUBSCRIBE";
    String ACK = "ACK";
    String NACK = "NACK";
    String BEGIN = "BEGIN";
    String COMMIT = "COMMIT";
    String ABORT = "ABORT";
    String DISCONNECT = "DISCONNECT";
    String MESSAGE = "MESSAGE";
    String RECEIPT = "RECEIPT";
    String ERROR = "ERROR";
    
    /** STOMP headers */
    String DESTINATION = "destination";
    String MESSAGE_ID = "message-id";
    String SUBSCRIPTION = "subscription";
    String RECEIPT_REQUESTED = "receipt";
    String TRANSACTION = "transaction";
    String ACK_MODE = "ack";
    String SELECTOR = "selector";
    String USER_ID = "login";
    String PASSWORD = "passcode";
    String CLIENT_ID = "client-id";
    
    /** Message transformation headers */
    String TRANSFORMATION = "transformation";
    String TRANSFORMATION_JSON = "jms-json";
    String TRANSFORMATION_XML = "jms-xml";
    String TRANSFORMATION_OBJECT_JSON = "jms-object-json";
    String TRANSFORMATION_OBJECT_XML = "jms-object-xml";
    String TRANSFORMATION_MAP_JSON = "jms-map-json";
    String TRANSFORMATION_MAP_XML = "jms-map-xml";
}

/**
 * STOMP transport implementation
 */
public class StompTransport extends TcpTransport {
    public StompTransport(WireFormat wireFormat, Socket socket) throws IOException;
    public StompTransport(WireFormat wireFormat, Socket socket, InitBuffer initBuffer) throws IOException;
}

/**
 * STOMP wire format for message serialization
 */
public class StompWireFormat implements WireFormat {
    /** Get maximum data length */
    public int getMaxDataLength();
    public void setMaxDataLength(int maxDataLength);
    
    /** Get maximum frame size */
    public long getMaxFrameSize();
    public void setMaxFrameSize(long maxFrameSize);
}

/**
 * STOMP protocol exception
 */
public class ProtocolException extends IOException {
    public ProtocolException(String message);
    public ProtocolException(String message, Throwable cause);
}

/**
 * AMQP transport implementation
 */
public class AmqpTransport extends TransportSupport {
    public AmqpTransport(WireFormat wireFormat, Socket socket) throws IOException;
    public AmqpTransport(WireFormat wireFormat, SSLSocket socket) throws IOException;
}

/**
 * AMQP NIO SSL transport
 */
public class AmqpNioSslTransport extends AmqpNioTransport {
    public AmqpNioSslTransport(WireFormat wireFormat, SocketChannel channel, 
                               SSLEngine engine) throws IOException;
}

/**
 * AMQP wire format implementation
 */
public class AmqpWireFormat implements WireFormat {
    /** AMQP version constants */
    public static final int DEFAULT_MAX_FRAME_SIZE = 1024 * 1024;
    public static final String QUEUE_PREFIX = "queue://";
    public static final String TOPIC_PREFIX = "topic://";
}

/**
 * AMQP utilities and support functions
 */
public class AmqpSupport {
    /** Convert JMS destination to AMQP address */
    public static String toAddress(Destination destination);
    
    /** Convert AMQP address to JMS destination */
    public static Destination toDestination(String address);
    
    /** Message property conversion */
    public static Object convertProperty(Object value);
}

/**
 * SASL authentication mechanism interface
 */
public interface SaslMechanism {
    /** Get mechanism name */
    String getName();
    
    /** Create initial response */
    byte[] getInitialResponse() throws SaslException;
    
    /** Process challenge response */
    byte[] getChallengeResponse(byte[] challenge) throws SaslException;
    
    /** Check if authentication is complete */
    boolean isComplete();
}

/**
 * MQTT transport implementation
 */
public class MQTTTransport extends TcpTransport {
    public MQTTTransport(WireFormat wireFormat, Socket socket) throws IOException;
    public MQTTTransport(WireFormat wireFormat, Socket socket, InitBuffer initBuffer) throws IOException;
}

/**
 * MQTT NIO SSL transport
 */
public class MQTTNIOSSLTransport extends MQTTNIOTransport {
    public MQTTNIOSSLTransport(WireFormat wireFormat, SocketChannel channel, 
                               SSLEngine engine) throws IOException;
}

/**
 * MQTT wire format for message serialization
 */
public class MQTTWireFormat implements WireFormat {
    /** MQTT protocol version */
    public void setVersion(int version);
    public int getVersion();
    
    /** Keep alive interval */
    public void setKeepAlive(int keepAlive);
    public int getKeepAlive();
}

/**
 * MQTT subscription management
 */
public class MQTTSubscription {
    /** Get topic filter */
    public String getTopicFilter();
    public void setTopicFilter(String topicFilter);
    
    /** Get quality of service level */
    public QoS getQoS();
    public void setQoS(QoS qos);
    
    /** MQTT QoS levels */
    public enum QoS {
        AT_MOST_ONCE(0),
        AT_LEAST_ONCE(1),
        EXACTLY_ONCE(2);
        
        private final int value;
        QoS(int value) { this.value = value; }
        public int getValue() { return value; }
    }
}

/**
 * MQTT protocol exception
 */
public class MQTTProtocolException extends IOException {
    public MQTTProtocolException(String message);
    public MQTTProtocolException(String message, Throwable cause);
}

/**
 * HTTP tunnel servlet for JMS over HTTP
 */
public class HttpTunnelServlet extends HttpServlet {
    /** Handle HTTP GET requests */
    protected void doGet(HttpServletRequest request, HttpServletResponse response) 
        throws ServletException, IOException;
    
    /** Handle HTTP POST requests */
    protected void doPost(HttpServletRequest request, HttpServletResponse response) 
        throws ServletException, IOException;
    
    /** Configure tunnel parameters */
    public void setMaxReadTimeoutMillis(long maxReadTimeoutMillis);
    public long getMaxReadTimeoutMillis();
}

/**
 * HTTP client transport implementation
 */
public class HttpClientTransport extends HttpTransportSupport {
    public HttpClientTransport(TextWireFormat wireFormat, URI uri) throws IOException;
    public HttpClientTransport(TextWireFormat wireFormat, URI uri, HttpClient httpClient) throws IOException;
}

/**
 * HTTP transport server
 */
public class HttpTransportServer extends TransportServerSupport {
    /** Set HTTP server configuration */
    public void setHttpServer(Server httpServer);
    public Server getHttpServer();
    
    /** Bind to address */
    public void bind() throws Exception;
    
    /** Start HTTP server */
    public void start() throws Exception;
    
    /** Stop HTTP server */
    public void stop() throws Exception;
}

/**
 * HTTP discovery agent for broker discovery
 */
public class HTTPDiscoveryAgent extends DiscoveryAgentSupport {
    /** Set registry URL */
    public void setRegistryURL(String registryURL);
    public String getRegistryURL();
    
    /** Start discovery */
    public void start() throws Exception;
    
    /** Stop discovery */
    public void stop() throws Exception;
    
    /** Advertise service */
    public void serviceFailed(DiscoveryEvent event) throws IOException;
    public void serviceAdded(DiscoveryEvent event);
    public void serviceRemoved(DiscoveryEvent event);
}

// STOMP transport configuration
BrokerService broker = new BrokerService();
broker.addConnector("stomp://localhost:61613");

// AMQP transport
broker.addConnector("amqp://localhost:5672");

// MQTT transport  
broker.addConnector("mqtt://localhost:1883");

// HTTP transport
broker.addConnector("http://localhost:8080");

// Multiple transports with SSL
broker.addConnector("stomp+ssl://localhost:61614");
broker.addConnector("mqtt+ssl://localhost:8883");

// Auto-detection transport (detects protocol automatically)
broker.addConnector("auto://localhost:5555");

/**
 * Wire format interface for protocol serialization
 */
public interface WireFormat {
    /** Marshal object to byte array */
    ByteSequence marshal(Object command) throws IOException;
    
    /** Unmarshal byte array to object */
    Object unmarshal(ByteSequence packet) throws IOException;
    
    /** Set version */
    void setVersion(int version);
    int getVersion();
}

/**
 * Text-based wire format for HTTP/WebSocket
 */
public interface TextWireFormat extends WireFormat {
    /** Marshal to text */
    String marshalText(Object command) throws IOException;
    
    /** Unmarshal from text */
    Object unmarshalText(String text) throws IOException;
}

/**
 * Discovery event for service discovery
 */
public class DiscoveryEvent {
    /** Get service name */
    public String getServiceName();
    
    /** Get service URI */
    public URI getUri();
    
    /** Check if service failed */
    public boolean isFailed();
}