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    }