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 */
017package org.apache.activemq.transport.stomp;
018
019import org.apache.activemq.transport.tcp.TcpTransport;
020import org.apache.activemq.util.ByteArrayOutputStream;
021import org.apache.activemq.util.DataByteArrayInputStream;
022
023import java.io.ByteArrayInputStream;
024import java.util.HashMap;
025
026public 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}