Author: jmesnil Date: 2010-01-21 09:41:55 -0500 (Thu, 21 Jan 2010) New Revision: 8826 Added: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompDestinationConverter.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompException.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java Removed: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ProtocolConverter.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/ProtocolException.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompDestinationConverter.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompException.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrame.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameDelimiter.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameError.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompMarshaller.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompPacketDecoder.java Modified: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ProtocolException.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrame.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameError.java branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompMarshaller.java Log: https://jira.jboss.org/jira/browse/HORNETQ-129: Implement STOMP v1.0 * sync with the trunk: svn merge -r 8806:8821 https://svn.jboss.org/repos/hornetq/trunk Copied: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp (from rev 8807, branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp) Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ProtocolConverter.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/ProtocolConverter.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ProtocolConverter.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,86 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -import java.util.Map; - -import org.hornetq.api.core.client.HornetQClient; -import org.hornetq.core.remoting.Packet; -import org.hornetq.core.remoting.impl.wireformat.CreateSessionMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionCloseMessage; -import org.hornetq.utils.UUIDGenerator; -import org.hornetq.utils.VersionLoader; - -/** - * A ProtocolConverter - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - * - * - */ -public class ProtocolConverter -{ - - public Packet toPacket(StompFrame frame) - { - String command = frame.getCommand(); - Map<String, Object> headers = frame.getHeaders(); - if (Stomp.Commands.CONNECT.equals(command)) - { - String login = (String)headers.get("login"); - String password = (String)headers.get("passcode"); - - String name = UUIDGenerator.getInstance().generateStringUUID(); - long sessionChannelID = 12; - return new CreateSessionMessage(name, - sessionChannelID, - VersionLoader.getVersion().getIncrementingVersion(), - login, - password, - HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, - false, - true, - true, - false, - HornetQClient.DEFAULT_CONFIRMATION_WINDOW_SIZE); - } - if (Stomp.Commands.DISCONNECT.equals(command)) - { - return new SessionCloseMessage(); - } - else - { - throw new RuntimeException("frame not supported: " + frame); - } - } - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Modified: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ProtocolException.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/ProtocolException.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/ProtocolException.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.hornetq.integration.stomp; +package org.hornetq.integration.protocol.stomp; import java.io.IOException; Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,124 +0,0 @@ -/** - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.hornetq.integration.stomp; - - -/** - * The standard verbs and headers used for the <a href="http://stomp.codehaus.org/">STOMP</a> protocol. - * - * @version $Revision: 57 $ - */ -public interface Stomp { - String NULL = "\u0000"; - String NEWLINE = "\n"; - - public static interface Commands { - String CONNECT = "CONNECT"; - String SEND = "SEND"; - String DISCONNECT = "DISCONNECT"; - String SUBSCRIBE = "SUB"; - String UNSUBSCRIBE = "UNSUB"; - String BEGIN_TRANSACTION = "BEGIN"; - String COMMIT_TRANSACTION = "COMMIT"; - String ABORT_TRANSACTION = "ABORT"; - String BEGIN = "BEGIN"; - String COMMIT = "COMMIT"; - String ABORT = "ABORT"; - String ACK = "ACK"; - } - - public interface Responses { - String CONNECTED = "CONNECTED"; - String ERROR = "ERROR"; - String MESSAGE = "MESSAGE"; - String RECEIPT = "RECEIPT"; - } - - public interface Headers { - String SEPERATOR = ":"; - String RECEIPT_REQUESTED = "receipt"; - String TRANSACTION = "transaction"; - String CONTENT_LENGTH = "content-length"; - - public interface Response { - String RECEIPT_ID = "receipt-id"; - } - - public interface Send { - String DESTINATION = "destination"; - String CORRELATION_ID = "correlation-id"; - String REPLY_TO = "reply-to"; - String EXPIRATION_TIME = "expires"; - String PRIORITY = "priority"; - String TYPE = "type"; - Object PERSISTENT = "persistent"; - } - - public interface Message { - String MESSAGE_ID = "message-id"; - String DESTINATION = "destination"; - String CORRELATION_ID = "correlation-id"; - String EXPIRATION_TIME = "expires"; - String REPLY_TO = "reply-to"; - String PRORITY = "priority"; - String REDELIVERED = "redelivered"; - String TIMESTAMP = "timestamp"; - String TYPE = "type"; - String SUBSCRIPTION = "subscription"; - } - - public interface Subscribe { - String DESTINATION = "destination"; - String ACK_MODE = "ack"; - String ID = "id"; - String SELECTOR = "selector"; - String DURABLE_SUBSCRIPTION_NAME = "durable-subscription-name"; - String NO_LOCAL = "no-local"; - - public interface AckModeValues { - String AUTO = "auto"; - String CLIENT = "client"; - } - } - - public interface Unsubscribe { - String DESTINATION = "destination"; - String ID = "id"; - } - - public interface Connect { - String LOGIN = "login"; - String PASSCODE = "passcode"; - String CLIENT_ID = "client-id"; - String REQUEST_ID = "request-id"; - } - - public interface Error { - String MESSAGE = "message"; - } - - public interface Connected { - String SESSION = "session"; - String RESPONSE_ID = "response-id"; - } - - public interface Ack { - String MESSAGE_ID = "message-id"; - } - } -} Copied: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java (from rev 8817, branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java) =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java (rev 0) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/Stomp.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -0,0 +1,124 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hornetq.integration.protocol.stomp; + + +/** + * The standard verbs and headers used for the <a href="http://stomp.codehaus.org/">STOMP</a> protocol. + * + * @version $Revision: 57 $ + */ +public interface Stomp { + String NULL = "\u0000"; + String NEWLINE = "\n"; + + public static interface Commands { + String CONNECT = "CONNECT"; + String SEND = "SEND"; + String DISCONNECT = "DISCONNECT"; + String SUBSCRIBE = "SUBSCRIBE"; + String UNSUBSCRIBE = "UNSUBSCRIBE"; + String BEGIN_TRANSACTION = "BEGIN"; + String COMMIT_TRANSACTION = "COMMIT"; + String ABORT_TRANSACTION = "ABORT"; + String BEGIN = "BEGIN"; + String COMMIT = "COMMIT"; + String ABORT = "ABORT"; + String ACK = "ACK"; + } + + public interface Responses { + String CONNECTED = "CONNECTED"; + String ERROR = "ERROR"; + String MESSAGE = "MESSAGE"; + String RECEIPT = "RECEIPT"; + } + + public interface Headers { + String SEPERATOR = ":"; + String RECEIPT_REQUESTED = "receipt"; + String TRANSACTION = "transaction"; + String CONTENT_LENGTH = "content-length"; + + public interface Response { + String RECEIPT_ID = "receipt-id"; + } + + public interface Send { + String DESTINATION = "destination"; + String CORRELATION_ID = "correlation-id"; + String REPLY_TO = "reply-to"; + String EXPIRATION_TIME = "expires"; + String PRIORITY = "priority"; + String TYPE = "type"; + Object PERSISTENT = "persistent"; + } + + public interface Message { + String MESSAGE_ID = "message-id"; + String DESTINATION = "destination"; + String CORRELATION_ID = "correlation-id"; + String EXPIRATION_TIME = "expires"; + String REPLY_TO = "reply-to"; + String PRORITY = "priority"; + String REDELIVERED = "redelivered"; + String TIMESTAMP = "timestamp"; + String TYPE = "type"; + String SUBSCRIPTION = "subscription"; + } + + public interface Subscribe { + String DESTINATION = "destination"; + String ACK_MODE = "ack"; + String ID = "id"; + String SELECTOR = "selector"; + String DURABLE_SUBSCRIPTION_NAME = "durable-subscription-name"; + String NO_LOCAL = "no-local"; + + public interface AckModeValues { + String AUTO = "auto"; + String CLIENT = "client"; + } + } + + public interface Unsubscribe { + String DESTINATION = "destination"; + String ID = "id"; + } + + public interface Connect { + String LOGIN = "login"; + String PASSCODE = "passcode"; + String CLIENT_ID = "client-id"; + String REQUEST_ID = "request-id"; + } + + public interface Error { + String MESSAGE = "message"; + } + + public interface Connected { + String SESSION = "session"; + String RESPONSE_ID = "response-id"; + } + + public interface Ack { + String MESSAGE_ID = "message-id"; + } + } +} Copied: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompDestinationConverter.java (from rev 8810, branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompDestinationConverter.java) =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompDestinationConverter.java (rev 0) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompDestinationConverter.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -0,0 +1,85 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.protocol.stomp; + +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.jms.client.HornetQQueue; +import org.hornetq.jms.client.HornetQTemporaryQueue; +import org.hornetq.jms.client.HornetQTemporaryTopic; +import org.hornetq.jms.client.HornetQTopic; + +/** + * A StompDestinationConverter + * + * @author jmesnil + * + * + */ +public class StompDestinationConverter +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + public static SimpleString convertDestination(String name) throws HornetQException + { + if (name == null) + { + throw new HornetQException(HornetQException.ILLEGAL_STATE, "No destination is specified!"); + } + else if (name.startsWith("/queue/")) + { + String queueName = name.substring("/queue/".length(), name.length()); + return HornetQQueue.createAddressFromName(queueName); + } + else if (name.startsWith("/topic/")) + { + String topicName = name.substring("/topic/".length(), name.length()); + return HornetQTopic.createAddressFromName(topicName); + } + else if (name.startsWith("/temp-queue/")) + { + String tempName = name.substring("/temp-queue/".length(), name.length()); + return HornetQTemporaryQueue.createAddressFromName(tempName); + } + else if (name.startsWith("/temp-topic/")) + { + String tempName = name.substring("/temp-topic/".length(), name.length()); + return HornetQTemporaryTopic.createAddressFromName(tempName); + } + else + { + throw new HornetQException(HornetQException.ILLEGAL_STATE, "Illegal destination name: [" + name + + "] -- StompConnect destinations " + + "must begine with one of: /queue/ /topic/ /temp-queue/ /temp-topic/"); + } + } + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Copied: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompException.java (from rev 8812, branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompException.java) =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompException.java (rev 0) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompException.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -0,0 +1,52 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.protocol.stomp; + +/** + * A StompException + * + * @author jmesnil + * + * + */ +public class StompException extends Exception +{ + + /** + * @param string + */ + public StompException(String string) + { + super(string); + } + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrame.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrame.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrame.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.hornetq.integration.stomp; +package org.hornetq.integration.protocol.stomp; import java.util.HashMap; import java.util.Map; Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameDelimiter.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,33 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -import org.jboss.netty.handler.codec.frame.DelimiterBasedFrameDecoder; -import org.jboss.netty.handler.codec.frame.Delimiters; - -/** - * A StompFrameDelimiter - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - */ -public class StompFrameDelimiter extends DelimiterBasedFrameDecoder -{ - - private static final int MAX_DATA_LENGTH = 1024 * 1024 * 100; - - public StompFrameDelimiter() - { - super(MAX_DATA_LENGTH, true, Delimiters.nulDelimiter()); - } -} Copied: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java (from rev 8812, branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameDelimiter.java) =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java (rev 0) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameDelimiter.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -0,0 +1,33 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.protocol.stomp; + +import org.jboss.netty.handler.codec.frame.DelimiterBasedFrameDecoder; +import org.jboss.netty.handler.codec.frame.Delimiters; + +/** + * A StompFrameDelimiter + * + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + */ +public class StompFrameDelimiter extends DelimiterBasedFrameDecoder +{ + + private static final int MAX_DATA_LENGTH = 1024 * 1024 * 100; + + public StompFrameDelimiter() + { + super(MAX_DATA_LENGTH, false, Delimiters.nulDelimiter()); + } +} Modified: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameError.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameError.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompFrameError.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.hornetq.integration.stomp; +package org.hornetq.integration.protocol.stomp; /** * Command indicating that an invalid Stomp Frame was received. Modified: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompMarshaller.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompMarshaller.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompMarshaller.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.hornetq.integration.stomp; +package org.hornetq.integration.protocol.stomp; import java.io.ByteArrayOutputStream; import java.io.DataOutput; Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompPacketDecoder.java 2010-01-19 16:04:31 UTC (rev 8807) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,52 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.core.remoting.Packet; -import org.hornetq.core.remoting.PacketDecoder; - -/** - * A StompPacketDecoder - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - */ -public class StompPacketDecoder implements PacketDecoder -{ - private final StompMarshaller marshaller = new StompMarshaller(); - - private final ProtocolConverter converter = new ProtocolConverter(); - - // PacketDecoder implementation ---------------------------------- - - public Packet decode(HornetQBuffer in) - { - StompFrame frame; - try - { - frame = marshaller.unmarshal(in); - System.out.println(">>> " + frame); - Packet packet = converter.toPacket(frame); - packet.setChannelID(1); - System.out.println(">>> " + packet); - - return packet; - } - catch (Exception e) - { - e.printStackTrace(); - return null; - } - } -} Copied: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java (from rev 8809, branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompPacketDecoder.java) =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java (rev 0) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/protocol/stomp/StompPacketDecoder.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -0,0 +1,51 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.protocol.stomp; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.buffers.impl.ChannelBufferWrapper; +import org.jboss.netty.buffer.ChannelBuffer; +import org.jboss.netty.channel.ChannelHandlerContext; +import org.jboss.netty.channel.ChannelPipelineCoverage; +import org.jboss.netty.channel.Channels; +import org.jboss.netty.channel.MessageEvent; +import org.jboss.netty.channel.SimpleChannelHandler; + +/** + * A StompPacketDecoder + * + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + */ +@ChannelPipelineCoverage("one") +public class StompPacketDecoder extends SimpleChannelHandler +{ + private final StompMarshaller marshaller; + + // PacketDecoder implementation ---------------------------------- + + public StompPacketDecoder(final StompMarshaller marshaller) + { + this.marshaller = marshaller; + } + + @Override + public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception + { + ChannelBuffer in = (ChannelBuffer)e.getMessage(); + HornetQBuffer buffer = new ChannelBufferWrapper(in); + StompFrame frame = marshaller.unmarshal(buffer); + + Channels.fireMessageReceived(ctx, frame); + } +} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/ProtocolException.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/ProtocolException.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/ProtocolException.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,50 +0,0 @@ -/** - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.hornetq.integration.stomp; - -import java.io.IOException; - -/** - * @author <a href="http://hiramchirino.com">chirino</a> - */ -public class ProtocolException extends IOException { - private static final long serialVersionUID = -2869735532997332242L; - private final boolean fatal; - - public ProtocolException() { - this(null); - } - - public ProtocolException(String s) { - this(s, false); - } - - public ProtocolException(String s, boolean fatal) { - this(s, fatal, null); - } - - public ProtocolException(String s, boolean fatal, Throwable cause) { - super(s); - this.fatal = fatal; - initCause(cause); - } - - public boolean isFatal() { - return fatal; - } -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,124 +0,0 @@ -/** - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.hornetq.integration.stomp; - - -/** - * The standard verbs and headers used for the <a href="http://stomp.codehaus.org/">STOMP</a> protocol. - * - * @version $Revision: 57 $ - */ -public interface Stomp { - String NULL = "\u0000"; - String NEWLINE = "\n"; - - public static interface Commands { - String CONNECT = "CONNECT"; - String SEND = "SEND"; - String DISCONNECT = "DISCONNECT"; - String SUBSCRIBE = "SUBSCRIBE"; - String UNSUBSCRIBE = "UNSUBSCRIBE"; - String BEGIN_TRANSACTION = "BEGIN"; - String COMMIT_TRANSACTION = "COMMIT"; - String ABORT_TRANSACTION = "ABORT"; - String BEGIN = "BEGIN"; - String COMMIT = "COMMIT"; - String ABORT = "ABORT"; - String ACK = "ACK"; - } - - public interface Responses { - String CONNECTED = "CONNECTED"; - String ERROR = "ERROR"; - String MESSAGE = "MESSAGE"; - String RECEIPT = "RECEIPT"; - } - - public interface Headers { - String SEPERATOR = ":"; - String RECEIPT_REQUESTED = "receipt"; - String TRANSACTION = "transaction"; - String CONTENT_LENGTH = "content-length"; - - public interface Response { - String RECEIPT_ID = "receipt-id"; - } - - public interface Send { - String DESTINATION = "destination"; - String CORRELATION_ID = "correlation-id"; - String REPLY_TO = "reply-to"; - String EXPIRATION_TIME = "expires"; - String PRIORITY = "priority"; - String TYPE = "type"; - Object PERSISTENT = "persistent"; - } - - public interface Message { - String MESSAGE_ID = "message-id"; - String DESTINATION = "destination"; - String CORRELATION_ID = "correlation-id"; - String EXPIRATION_TIME = "expires"; - String REPLY_TO = "reply-to"; - String PRORITY = "priority"; - String REDELIVERED = "redelivered"; - String TIMESTAMP = "timestamp"; - String TYPE = "type"; - String SUBSCRIPTION = "subscription"; - } - - public interface Subscribe { - String DESTINATION = "destination"; - String ACK_MODE = "ack"; - String ID = "id"; - String SELECTOR = "selector"; - String DURABLE_SUBSCRIPTION_NAME = "durable-subscription-name"; - String NO_LOCAL = "no-local"; - - public interface AckModeValues { - String AUTO = "auto"; - String CLIENT = "client"; - } - } - - public interface Unsubscribe { - String DESTINATION = "destination"; - String ID = "id"; - } - - public interface Connect { - String LOGIN = "login"; - String PASSCODE = "passcode"; - String CLIENT_ID = "client-id"; - String REQUEST_ID = "request-id"; - } - - public interface Error { - String MESSAGE = "message"; - } - - public interface Connected { - String SESSION = "session"; - String RESPONSE_ID = "response-id"; - } - - public interface Ack { - String MESSAGE_ID = "message-id"; - } - } -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompDestinationConverter.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompDestinationConverter.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompDestinationConverter.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,85 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.SimpleString; -import org.hornetq.jms.client.HornetQQueue; -import org.hornetq.jms.client.HornetQTemporaryQueue; -import org.hornetq.jms.client.HornetQTemporaryTopic; -import org.hornetq.jms.client.HornetQTopic; - -/** - * A StompDestinationConverter - * - * @author jmesnil - * - * - */ -public class StompDestinationConverter -{ - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - public static SimpleString convertDestination(String name) throws HornetQException - { - if (name == null) - { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "No destination is specified!"); - } - else if (name.startsWith("/queue/")) - { - String queueName = name.substring("/queue/".length(), name.length()); - return HornetQQueue.createAddressFromName(queueName); - } - else if (name.startsWith("/topic/")) - { - String topicName = name.substring("/topic/".length(), name.length()); - return HornetQTopic.createAddressFromName(topicName); - } - else if (name.startsWith("/temp-queue/")) - { - String tempName = name.substring("/temp-queue/".length(), name.length()); - return HornetQTemporaryQueue.createAddressFromName(tempName); - } - else if (name.startsWith("/temp-topic/")) - { - String tempName = name.substring("/temp-topic/".length(), name.length()); - return HornetQTemporaryTopic.createAddressFromName(tempName); - } - else - { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "Illegal destination name: [" + name + - "] -- StompConnect destinations " + - "must begine with one of: /queue/ /topic/ /temp-queue/ /temp-topic/"); - } - } - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompException.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompException.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompException.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,52 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -/** - * A StompException - * - * @author jmesnil - * - * - */ -public class StompException extends Exception -{ - - /** - * @param string - */ - public StompException(String string) - { - super(string); - } - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrame.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrame.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrame.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,70 +0,0 @@ -/** - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.hornetq.integration.stomp; - -import java.util.HashMap; -import java.util.Map; - -/** - * Represents all the data in a STOMP frame. - * - * @author <a href="http://hiramchirino.com">chirino</a> - */ -public class StompFrame -{ - private static final byte[] NO_DATA = new byte[] {}; - - private String command; - - private Map<String, Object> headers; - - private byte[] content = StompFrame.NO_DATA; - - public StompFrame() - { - this.headers = new HashMap<String, Object>(); - } - - public StompFrame(String command, Map<String, Object> headers, byte[] data) - { - this.command = command; - this.headers = headers; - this.content = data; - } - - public String getCommand() - { - return command; - } - - public byte[] getContent() - { - return content; - } - - public Map<String, Object> getHeaders() - { - return headers; - } - - @Override - public String toString() - { - return "StompFrame[command=" + command + ", headers=" + headers + ",content-length=" + content.length + "]"; - } -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameDelimiter.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameDelimiter.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameDelimiter.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,33 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -import org.jboss.netty.handler.codec.frame.DelimiterBasedFrameDecoder; -import org.jboss.netty.handler.codec.frame.Delimiters; - -/** - * A StompFrameDelimiter - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - */ -public class StompFrameDelimiter extends DelimiterBasedFrameDecoder -{ - - private static final int MAX_DATA_LENGTH = 1024 * 1024 * 100; - - public StompFrameDelimiter() - { - super(MAX_DATA_LENGTH, false, Delimiters.nulDelimiter()); - } -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameError.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameError.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompFrameError.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,35 +0,0 @@ -/** - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.hornetq.integration.stomp; - -/** - * Command indicating that an invalid Stomp Frame was received. - * - * @author <a href="http://hiramchirino.com">chirino</a> - */ -public class StompFrameError extends StompFrame { - private final ProtocolException exception; - - public StompFrameError(ProtocolException exception) { - this.exception = exception; - } - - public ProtocolException getException() { - return exception; - } -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompMarshaller.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompMarshaller.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompMarshaller.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,191 +0,0 @@ -/** - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.hornetq.integration.stomp; - -import java.io.ByteArrayOutputStream; -import java.io.DataOutput; -import java.io.DataOutputStream; -import java.io.IOException; -import java.util.HashMap; -import java.util.Iterator; -import java.util.Map; - -import org.hornetq.api.core.HornetQBuffer; - -/** - * Implements marshalling and unmarsalling the <a href="http://stomp.codehaus.org/">Stomp</a> protocol. - */ -public class StompMarshaller { - private static final byte[] NO_DATA = new byte[]{}; - private static final byte[] END_OF_FRAME = new byte[]{0, '\n'}; - private static final int MAX_COMMAND_LENGTH = 1024; - private static final int MAX_HEADER_LENGTH = 1024 * 10; - private static final int MAX_HEADERS = 1000; - private static final int MAX_DATA_LENGTH = 1024 * 1024 * 100; - private int version = 1; - - public int getVersion() { - return version; - } - - public void setVersion(int version) { - this.version = version; - } - - public byte[] marshal(StompFrame command) throws IOException { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - DataOutputStream dos = new DataOutputStream(baos); - marshal(command, dos); - dos.close(); - return baos.toByteArray(); - } - - public void marshal(StompFrame stomp, DataOutput os) throws IOException { - StringBuffer buffer = new StringBuffer(); - buffer.append(stomp.getCommand()); - buffer.append(Stomp.NEWLINE); - - // Output the headers. - for (Iterator iter = stomp.getHeaders().entrySet().iterator(); iter.hasNext();) { - Map.Entry entry = (Map.Entry) iter.next(); - buffer.append(entry.getKey()); - buffer.append(Stomp.Headers.SEPERATOR); - buffer.append(entry.getValue()); - buffer.append(Stomp.NEWLINE); - } - - // Add a newline to seperate the headers from the content. - buffer.append(Stomp.NEWLINE); - - os.write(buffer.toString().getBytes("UTF-8")); - os.write(stomp.getContent()); - os.write(END_OF_FRAME); - } - - public StompFrame unmarshal(HornetQBuffer in) throws IOException { - - try { - String action = null; - - // skip white space to next real action line - while (true) { - action = readLine(in, MAX_COMMAND_LENGTH, "The maximum command length was exceeded"); - if (action == null) { - throw new IOException("connection was closed"); - } - else { - action = action.trim(); - if (action.length() > 0) { - break; - } - } - } - - // Parse the headers - HashMap headers = new HashMap(25); - while (true) { - String line = readLine(in, MAX_HEADER_LENGTH, "The maximum header length was exceeded"); - if (line != null && line.trim().length() > 0) { - - if (headers.size() > MAX_HEADERS) { - throw new ProtocolException("The maximum number of headers was exceeded", true); - } - - try { - int seperator_index = line.indexOf(Stomp.Headers.SEPERATOR); - String name = line.substring(0, seperator_index).trim(); - String value = line.substring(seperator_index + 1, line.length()).trim(); - headers.put(name, value); - } - catch (Exception e) { - throw new ProtocolException("Unable to parser header line [" + line + "]", true); - } - } - else { - break; - } - } - - // Read in the data part. - byte[] data = NO_DATA; - String contentLength = (String) headers.get(Stomp.Headers.CONTENT_LENGTH); - if (contentLength != null) { - - // Bless the client, he's telling us how much data to read in. - int length; - try { - length = Integer.parseInt(contentLength.trim()); - } - catch (NumberFormatException e) { - throw new ProtocolException("Specified content-length is not a valid integer", true); - } - - if (length > MAX_DATA_LENGTH) { - throw new ProtocolException("The maximum data length was exceeded", true); - } - - data = new byte[length]; - in.readBytes(data); - - if (in.readByte() != 0) { - throw new ProtocolException(Stomp.Headers.CONTENT_LENGTH + " bytes were read and " + "there was no trailing null byte", true); - } - } - else { - - // We don't know how much to read.. data ends when we hit a 0 - byte b; - ByteArrayOutputStream baos = null; - while (in.readableBytes() > 0 && (b = in.readByte()) != 0) { - - if (baos == null) { - baos = new ByteArrayOutputStream(); - } - else if (baos.size() > MAX_DATA_LENGTH) { - throw new ProtocolException("The maximum data length was exceeded", true); - } - - baos.write(b); - } - - if (baos != null) { - baos.close(); - data = baos.toByteArray(); - } - } - - return new StompFrame(action, headers, data); - } - catch (ProtocolException e) { - return new StompFrameError(e); - } - } - - protected String readLine(HornetQBuffer in, int maxLength, String errorMessage) throws IOException { - byte b; - ByteArrayOutputStream baos = new ByteArrayOutputStream(maxLength); - while ((b = in.readByte()) != '\n') { - if (baos.size() > maxLength) { - throw new ProtocolException(errorMessage, true); - } - baos.write(b); - } - byte[] sequence = baos.toByteArray(); - return new String(sequence, "UTF-8"); - } -} Deleted: branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompPacketDecoder.java =================================================================== --- branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompPacketDecoder.java 2010-01-21 14:40:58 UTC (rev 8825) +++ branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/StompPacketDecoder.java 2010-01-21 14:41:55 UTC (rev 8826) @@ -1,51 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.stomp; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.core.buffers.impl.ChannelBufferWrapper; -import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ChannelPipelineCoverage; -import org.jboss.netty.channel.Channels; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelHandler; - -/** - * A StompPacketDecoder - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - */ -@ChannelPipelineCoverage("one") -public class StompPacketDecoder extends SimpleChannelHandler -{ - private final StompMarshaller marshaller; - - // PacketDecoder implementation ---------------------------------- - - public StompPacketDecoder(final StompMarshaller marshaller) - { - this.marshaller = marshaller; - } - - @Override - public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception - { - ChannelBuffer in = (ChannelBuffer)e.getMessage(); - HornetQBuffer buffer = new ChannelBufferWrapper(in); - StompFrame frame = marshaller.unmarshal(buffer); - - Channels.fireMessageReceived(ctx, frame); - } -}