Better protocol error handling.

Fixed http://issues.apache.org/activemq/browse/AMQ-649



git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@418602 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/ProtocolConverter.java

@@ -19,7 +19,6 @@ package org.apache.activemq.transport.stomp2;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.io.PrintWriter;
-import java.net.ProtocolException;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -87,13 +86,13 @@ public class ProtocolConverter {
     	}
     }
 
-    protected ResponseHandler createResponseHandler(StompCommand command){
+    protected ResponseHandler createResponseHandler(StompFrame command){
         final String receiptId = (String) command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED);
         // A response may not be needed.
         if( receiptId != null ) {
 	        return new ResponseHandler() {
 	    		public void onResponse(ProtocolConverter converter, Response response) throws IOException {
-	                StompCommand sc = new StompCommand();
+	                StompFrame sc = new StompFrame();
 	                sc.setHeaders(new HashMap(5));
 	                sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId);
 	        		transportFilter.sendToStomp(sc);
@@ -112,7 +111,7 @@ public class ProtocolConverter {
 		transportFilter.sendToActiveMQ(command);
 	}
 
-	protected void sendToStomp(StompCommand command) throws IOException {
+	protected void sendToStomp(StompFrame command) throws IOException {
 		transportFilter.sendToStomp(command);
 	}
 
@@ -120,9 +119,13 @@ public class ProtocolConverter {
      * Convert a stomp command
      * @param command
      */
-	public void onStompCommad( StompCommand command ) throws IOException, JMSException {
+	public void onStompCommad( StompFrame command ) throws IOException, JMSException {
 		try {
 			
+			if( command.getClass() == StompFrameError.class ) {
+				throw ((StompFrameError)command).getException();
+			}
+			
 			String action = command.getAction();
 	        if (action.startsWith(Stomp.Commands.SEND))
 	            onStompSend(command);
@@ -161,13 +164,15 @@ public class ProtocolConverter {
             	headers.put(Stomp.Headers.Response.RECEIPT_ID, receiptId);
             }
         	
-        	StompCommand errorMessage = new StompCommand(Stomp.Responses.ERROR,headers,baos.toByteArray());
+        	StompFrame errorMessage = new StompFrame(Stomp.Responses.ERROR,headers,baos.toByteArray());
 			sendToStomp(errorMessage);
 			
+			if( e.isFatal() )
+				getTransportFilter().onException(e);
         }
 	}
 	
-	protected void onStompSend(StompCommand command) throws IOException, JMSException {
+	protected void onStompSend(StompFrame command) throws IOException, JMSException {
 		checkConnected();
 
     	Map headers = command.getHeaders();
@@ -193,7 +198,7 @@ public class ProtocolConverter {
 	}
 	
 
-    protected void onStompAck(StompCommand command) throws ProtocolException {
+    protected void onStompAck(StompFrame command) throws ProtocolException {
 		checkConnected();
 
     	// TODO: acking with just a message id is very bogus
@@ -231,7 +236,7 @@ public class ProtocolConverter {
 	}
     
 
-	protected void onStompBegin(StompCommand command) throws ProtocolException {
+	protected void onStompBegin(StompFrame command) throws ProtocolException {
 		checkConnected();
 
 		Map headers = command.getHeaders();
@@ -258,7 +263,7 @@ public class ProtocolConverter {
 		
 	}
 	
-	protected void onStompCommit(StompCommand command) throws ProtocolException {
+	protected void onStompCommit(StompFrame command) throws ProtocolException {
 		checkConnected();
 
 		Map headers = command.getHeaders();
@@ -283,7 +288,7 @@ public class ProtocolConverter {
 		sendToActiveMQ(tx, createResponseHandler(command));
 	}
 
-	protected void onStompAbort(StompCommand command) throws ProtocolException {
+	protected void onStompAbort(StompFrame command) throws ProtocolException {
 		checkConnected();
     	Map headers = command.getHeaders();
 		
@@ -308,7 +313,7 @@ public class ProtocolConverter {
 		
 	}
 
-	protected void onStompSubscribe(StompCommand command) throws ProtocolException {
+	protected void onStompSubscribe(StompFrame command) throws ProtocolException {
 		checkConnected();
     	Map headers = command.getHeaders();
         
@@ -343,7 +348,7 @@ public class ProtocolConverter {
 		
 	}
 
-	protected void onStompUnsubscribe(StompCommand command) throws ProtocolException {
+	protected void onStompUnsubscribe(StompFrame command) throws ProtocolException {
 		checkConnected();
     	Map headers = command.getHeaders();
 
@@ -375,7 +380,7 @@ public class ProtocolConverter {
         throw new ProtocolException("No subscription matched.");
 	}
 
-	protected void onStompConnect(StompCommand command) throws ProtocolException {
+	protected void onStompConnect(StompFrame command) throws ProtocolException {
 
 		if(connected.get()) {
 			throw new ProtocolException("Allready connected.");
@@ -422,7 +427,7 @@ public class ProtocolConverter {
 		                    responseHeaders.put(Stomp.Headers.Connected.RESPONSE_ID, requestId);
 	            		}
 	                    
-	                    StompCommand sc = new StompCommand();
+	                    StompFrame sc = new StompFrame();
 	                    sc.setAction(Stomp.Responses.CONNECTED);
 	                    sc.setHeaders(responseHeaders);
 	                    sendToStomp(sc);
@@ -434,7 +439,7 @@ public class ProtocolConverter {
 		
 	}
 
-	protected void onStompDisconnect(StompCommand command) throws ProtocolException {
+	protected void onStompDisconnect(StompFrame command) throws ProtocolException {
 		checkConnected();
 		sendToActiveMQ(new ShutdownInfo(), createResponseHandler(command));
 		connected.set(false);
@@ -473,7 +478,7 @@ public class ProtocolConverter {
 		
 	}
 
-	public  ActiveMQMessage convertMessage(StompCommand command) throws IOException, JMSException {
+	public  ActiveMQMessage convertMessage(StompFrame command) throws IOException, JMSException {
 		Map headers = command.getHeaders();
         
         // now the body
@@ -488,7 +493,7 @@ public class ProtocolConverter {
             try {
 				text.setText(new String(command.getContent(), "UTF-8"));
 			} catch (Throwable e) {
-				throw (ProtocolException)new ProtocolException("Text could not bet set: "+e).initCause(e);
+				throw new ProtocolException("Text could not bet set: "+e, false, e);
 			}
             msg = text;
         }
@@ -530,9 +535,9 @@ public class ProtocolConverter {
         return msg;        
 	}
 	
-	public StompCommand convertMessage(ActiveMQMessage message) throws IOException, JMSException {
+	public StompFrame convertMessage(ActiveMQMessage message) throws IOException, JMSException {
 
-		StompCommand command = new StompCommand();
+		StompFrame command = new StompFrame();
 		command.setAction(Stomp.Responses.MESSAGE);
 		
 		HashMap headers = new HashMap();
@@ -620,8 +625,4 @@ public class ProtocolConverter {
 		this.transportFilter = transportFilter;
 	}
 	
-	public void onStompExcepton(IOException error) {
-		// TODO Auto-generated method stub
-	}
-
 }

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/ProtocolException.java

@@ -0,0 +1,50 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.apache.activemq.transport.stomp2;
+
+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;
+	}
+
+}

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/StompFrame.java

@@ -30,7 +30,7 @@ import org.apache.activemq.state.CommandVisitor;
  * 
  * @author <a href="http://hiramchirino.com">chirino</a> 
  */
-public class StompCommand implements Command {
+public class StompFrame implements Command {
 
     private static final byte[] NO_DATA = new byte[]{};
 
@@ -38,13 +38,13 @@ public class StompCommand implements Command {
 	private Map headers = Collections.EMPTY_MAP;
 	private byte[] content = NO_DATA;
 
-	public StompCommand(String command, HashMap headers, byte[] data) {
+	public StompFrame(String command, HashMap headers, byte[] data) {
 		this.action = command;
 		this.headers = headers;
 		this.content = data;
 	}
 
-	public StompCommand() {
+	public StompFrame() {
 	}
 
 	public String getAction() {

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/StompFrameError.java

@@ -0,0 +1,38 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.apache.activemq.transport.stomp2;
+
+/**
+ * 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;
+	}
+
+
+}

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/StompSubscription.java

@@ -69,7 +69,7 @@ public class StompSubscription {
             protocolConverter.getTransportFilter().sendToActiveMQ(ack);
         }
         
-        StompCommand command = protocolConverter.convertMessage(message);
+        StompFrame command = protocolConverter.convertMessage(message);
         
         command.setAction(Stomp.Responses.MESSAGE);        
         if (subscriptionId!=null) {

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/StompTransportFilter.java

@@ -46,14 +46,6 @@ public class StompTransportFilter extends TransportFilter {
 		protocolConverter.setTransportFilter(this);
 	}
 
-	public void start() throws Exception {
-		super.start();
-	}
-	
-	public void stop() throws Exception {
-		super.stop();
-	}
-	
 	public void oneway(Command command) throws IOException {
         try {
         	protocolConverter.onActiveMQCommad(command);
@@ -64,7 +56,7 @@ public class StompTransportFilter extends TransportFilter {
 	
 	public void onCommand(Command command) {
         try {
-        	protocolConverter.onStompCommad((StompCommand) command);
+        	protocolConverter.onStompCommad((StompFrame) command);
 		} catch (IOException e) {
 			onException(e);
 		} catch (JMSException e) {
@@ -72,24 +64,16 @@ public class StompTransportFilter extends TransportFilter {
 		}
 	}
 	
-	public void onException(IOException error) {
-		protocolConverter.onStompExcepton(error);
-		transportListener.onException(error);
-	}
-
-
 	public void sendToActiveMQ(Command command) {
 		synchronized(sendToActiveMQMutex) {
 			transportListener.onCommand(command);
 		}
 	}
 	
-	public void sendToStomp(StompCommand command) throws IOException {
+	public void sendToStomp(StompFrame command) throws IOException {
 		synchronized(sendToStompMutex) {
 			next.oneway(command);
 		}
 	}
 
-
-	
 }

activemq-core/src/main/java/org/apache/activemq/transport/stomp2/StompWireFormat.java

@@ -19,7 +19,6 @@ package org.apache.activemq.transport.stomp2;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
-import java.net.ProtocolException;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -62,7 +61,7 @@ public class StompWireFormat implements WireFormat {
     }
 
     public void marshal(Object command, DataOutputStream os) throws IOException {
-		StompCommand stomp = (org.apache.activemq.transport.stomp2.StompCommand) command;
+		StompFrame stomp = (org.apache.activemq.transport.stomp2.StompFrame) command;
 
 		StringBuffer buffer = new StringBuffer();
 		buffer.append(stomp.getAction());
@@ -88,6 +87,7 @@ public class StompWireFormat implements WireFormat {
 
     public Object unmarshal(DataInputStream in) throws IOException {
         	
+        try {
 			String action = null;
 			
 			// skip white space to next real action line
@@ -110,7 +110,7 @@ public class StompWireFormat implements WireFormat {
 			    if (line != null && line.trim().length() > 0) {
 			    	
 			    	if( headers.size() > MAX_HEADERS )
-            		throw new ProtocolException("The maximum number of headers was exceeded");
+			    		throw new ProtocolException("The maximum number of headers was exceeded", true);
 			    	
 			        try {
 			            int seperator_index = line.indexOf(Stomp.Headers.SEPERATOR);
@@ -119,7 +119,7 @@ public class StompWireFormat implements WireFormat {
 			            headers.put(name, value);
 			        }
 			        catch (Exception e) {
-                    throw new ProtocolException("Unable to parser header line [" + line + "]");
+			            throw new ProtocolException("Unable to parser header line [" + line + "]", true);
 			        }
 			    }
 			    else {
@@ -137,17 +137,17 @@ public class StompWireFormat implements WireFormat {
 				try {
 					length = Integer.parseInt(contentLength.trim());
 				} catch (NumberFormatException e) {
-				throw new ProtocolException("Specified content-length is not a valid integer");
+					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");
+					throw new ProtocolException("The maximum data length was exceeded", true);
 				
 			    data = new byte[length];
 			    in.readFully(data);
 			    
 			    if (in.readByte() != 0) {
-                throw new ProtocolException(Stomp.Headers.CONTENT_LENGTH+" bytes were read and " + "there was no trailing null byte");
+			        throw new ProtocolException(Stomp.Headers.CONTENT_LENGTH+" bytes were read and " + "there was no trailing null byte", true);
 			    }
 			
 			} else {
@@ -160,7 +160,7 @@ public class StompWireFormat implements WireFormat {
 					if( baos == null ) {
 			    		baos = new ByteArrayOutputStream();
 			    	} else if( baos.size() > MAX_DATA_LENGTH ) {
-            		throw new ProtocolException("The maximum data length was exceeded");
+			    		throw new ProtocolException("The maximum data length was exceeded", true);
 			    	}
 			    
 			        baos.write(b);
@@ -173,7 +173,11 @@ public class StompWireFormat implements WireFormat {
 			    
 			}
 			
-        return new StompCommand(action, headers, data); 
+			return new StompFrame(action, headers, data);
+			
+		} catch (ProtocolException e) {
+			return new StompFrameError(e);
+		} 
 
     }
 
@@ -182,7 +186,7 @@ public class StompWireFormat implements WireFormat {
         ByteArrayOutputStream baos=new ByteArrayOutputStream(maxLength);
         while ((b = in.readByte()) != '\n') {
         	if( baos.size() > maxLength )
-        		throw new ProtocolException(errorMessage);
+        		throw new ProtocolException(errorMessage, true);
             baos.write(b);
         }
         ByteSequence sequence = baos.toByteSequence();