001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.activemq.transport.stomp;
018
019 import org.apache.activemq.transport.tcp.TcpTransport;
020 import org.apache.activemq.util.ByteArrayOutputStream;
021 import org.apache.activemq.util.DataByteArrayInputStream;
022
023 import java.io.ByteArrayInputStream;
024 import java.util.HashMap;
025
026 public class StompCodec {
027
028 final static byte[] crlfcrlf = new byte[]{'\r','\n','\r','\n'};
029 TcpTransport transport;
030
031 ByteArrayOutputStream currentCommand = new ByteArrayOutputStream();
032 boolean processedHeaders = false;
033 String action;
034 HashMap<String, String> headers;
035 int contentLength = -1;
036 int readLength = 0;
037 int previousByte = -1;
038
039 public StompCodec(TcpTransport transport) {
040 this.transport = transport;
041 }
042
043 public void parse(ByteArrayInputStream input, int readSize) throws Exception {
044 int i = 0;
045 int b;
046 while(i++ < readSize) {
047 b = input.read();
048 // skip repeating nulls
049 if (!processedHeaders && previousByte == 0 && b == 0) {
050 continue;
051 }
052
053 if (!processedHeaders) {
054 currentCommand.write(b);
055 // end of headers section, parse action and header
056 if (b == '\n' && (previousByte == '\n' || currentCommand.endsWith(crlfcrlf))) {
057 if (transport.getWireFormat() instanceof StompWireFormat) {
058 DataByteArrayInputStream data = new DataByteArrayInputStream(currentCommand.toByteArray());
059 action = ((StompWireFormat)transport.getWireFormat()).parseAction(data);
060 headers = ((StompWireFormat)transport.getWireFormat()).parseHeaders(data);
061 String contentLengthHeader = headers.get(Stomp.Headers.CONTENT_LENGTH);
062 if ((action.equals(Stomp.Commands.SEND) || action.equals(Stomp.Responses.MESSAGE)) && contentLengthHeader != null) {
063 contentLength = ((StompWireFormat)transport.getWireFormat()).parseContentLength(contentLengthHeader);
064 } else {
065 contentLength = -1;
066 }
067 }
068 processedHeaders = true;
069 currentCommand.reset();
070 }
071 } else {
072
073 if (contentLength == -1) {
074 // end of command reached, unmarshal
075 if (b == 0) {
076 processCommand();
077 } else {
078 currentCommand.write(b);
079 }
080 } else {
081 // read desired content length
082 if (readLength++ == contentLength) {
083 processCommand();
084 readLength = 0;
085 } else {
086 currentCommand.write(b);
087 }
088 }
089 }
090
091 previousByte = b;
092 }
093 }
094
095 protected void processCommand() throws Exception {
096 StompFrame frame = new StompFrame(action, headers, currentCommand.toByteArray());
097 transport.doConsume(frame);
098 processedHeaders = false;
099 currentCommand.reset();
100 contentLength = -1;
101 }
102 }