src/main/csharp/Transport/Stomp/StompWireFormat.cs - activemq-nms-openwire - Git at Google

 /*
  * 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.
  */
 using Apache.NMS.ActiveMQ.Commands;
 using Apache.NMS.ActiveMQ.Transport;
 using Apache.NMS;
 using System;
 using System.Collections;
 using System.IO;
 using System.Text;

 namespace Apache.NMS.ActiveMQ.Transport.Stomp
 {
     /// <summary>
     /// Implements the <a href="http://stomp.codehaus.org/">STOMP</a> protocol.
     /// </summary>
     public class StompWireFormat : IWireFormat
     {
         private Encoding encoding = new UTF8Encoding();
         private ITransport transport;
         private IDictionary consumers = Hashtable.Synchronized(new Hashtable());

         public StompWireFormat()
         {
         }

         public ITransport Transport {
             get { return transport; }
             set { transport = value; }
         }

         public int Version {
             get { return 1; }
         }

         public void Marshal(Object o, BinaryWriter binaryWriter)
         {
             Tracer.Debug(">>>> " + o);
             StompFrameStream ds = new StompFrameStream(binaryWriter, encoding);

             if (o is ConnectionInfo)
             {
                 WriteConnectionInfo((ConnectionInfo) o, ds);
             }
             else if (o is ActiveMQMessage)
             {
                 WriteMessage((ActiveMQMessage) o, ds);
             }
             else if (o is ConsumerInfo)
             {
                 WriteConsumerInfo((ConsumerInfo) o, ds);
             }
             else if (o is MessageAck)
             {
                 WriteMessageAck((MessageAck) o, ds);
             }
             else if (o is TransactionInfo)
             {
                 WriteTransactionInfo((TransactionInfo) o, ds);
             }
             else if (o is ShutdownInfo)
             {
                 WriteShutdownInfo((ShutdownInfo) o, ds);
             }
             else if (o is RemoveInfo)
             {
                 WriteRemoveInfo((RemoveInfo) o, ds);
             }
             else if (o is Command)
             {
                 Command command = o as Command;
                 if (command.ResponseRequired)
                 {
                     Response response = new Response();
                     response.CorrelationId = command.CommandId;
                     SendCommand(response);
                     Tracer.Debug("#### Autorespond to command: " + o.GetType());
                 }
             }
             else
             {
                 Tracer.Debug("#### Ignored command: " + o.GetType());
             }
         }


         internal String ReadLine(BinaryReader dis)
         {
             MemoryStream ms = new MemoryStream();
             while (true)
             {
                 int nextChar = dis.Read();
                 if (nextChar < 0)
                 {
                     throw new IOException("Peer closed the stream.");
                 }
                 if( nextChar == 10 )
                 {
                     break;
                 }
                 ms.WriteByte((byte)nextChar);
             }
             byte[] data = ms.ToArray();
             return encoding.GetString(data, 0, data.Length);
         }

         public Object Unmarshal(BinaryReader dis)
         {
             string command;
             do {
                 command = ReadLine(dis);
             }
             while (command == "");

             Tracer.Debug("<<<< command: " + command);

             IDictionary headers = new Hashtable();
             string line;
             while ((line = ReadLine(dis)) != "")
             {
                 int idx = line.IndexOf(':');
                 if (idx > 0)
                 {
                     string key = line.Substring(0, idx);
                     string value = line.Substring(idx + 1);
                     headers[key] = value;

                     Tracer.Debug("<<<< header: " + key + " = " + value);
                 }
                 else
                 {
                     // lets ignore this bad header!
                 }
             }
             byte[] content = null;
             string length = ToString(headers["content-length"]);
             if (length != null)
             {
                 int size = Int32.Parse(length);
                 content = dis.ReadBytes(size);
                 // Read the terminating NULL byte for this frame.
                 int nullByte = dis.Read();
                 if(nullByte != 0)
                 {
                     Tracer.Debug("<<<< error reading frame null byte.");
                 }
             }
             else
             {
                 MemoryStream ms = new MemoryStream();
                 int nextChar;
                 while((nextChar = dis.Read()) != 0)
                 {
                     if( nextChar < 0 )
                     {
                         // EOF ??
                         break;
                     }
                     ms.WriteByte((byte)nextChar);
                 }
                 content = ms.ToArray();
             }
             Object answer = CreateCommand(command, headers, content);
             Tracer.Debug("<<<< received: " + answer);
             return answer;
         }

         protected virtual Object CreateCommand(string command, IDictionary headers, byte[] content)
         {
             if(command == "RECEIPT" || command == "CONNECTED")
             {
                 string text = RemoveHeader(headers, "receipt-id");
                 if(text != null)
                 {
                     Response answer = new Response();
                     if(text.StartsWith("ignore:"))
                     {
                         text = text.Substring("ignore:".Length);
                     }

                     answer.CorrelationId = Int32.Parse(text);
                     return answer;
                 }
                 else if(command == "CONNECTED")
                 {
                     text = RemoveHeader(headers, "response-id");
                     if (text != null)
                     {
                         Response answer = new Response();
                         answer.CorrelationId = Int32.Parse(text);
                         return answer;
                     }
                 }
             }
             else if(command == "ERROR")
             {
                 string text = RemoveHeader(headers, "receipt-id");

                 if(text != null && text.StartsWith("ignore:"))
                 {
                     Response answer = new Response();
                     answer.CorrelationId = Int32.Parse(text.Substring("ignore:".Length));
                     return answer;
                 }
                 else
                 {
                     ExceptionResponse answer = new ExceptionResponse();
                     if(text != null)
                     {
                         answer.CorrelationId = Int32.Parse(text);
                     }

                     BrokerError error = new BrokerError();
                     error.Message = RemoveHeader(headers, "message");
                     error.ExceptionClass = RemoveHeader(headers, "exceptionClass");
                     // TODO is this the right header?
                     answer.Exception = error;
                     return answer;
                 }
             }
             else if (command == "MESSAGE")
             {
                 return ReadMessage(command, headers, content);
             }
             Tracer.Error("Unknown command: " + command + " headers: " + headers);
             return null;
         }

         protected virtual Command ReadMessage(string command, IDictionary headers, byte[] content)
         {
             ActiveMQMessage message = null;
             if (headers.Contains("content-length"))
             {
                 message = new ActiveMQBytesMessage();
                 message.Content = content;
             }
             else
             {
                 message = new ActiveMQTextMessage(encoding.GetString(content, 0, content.Length));
             }

             // TODO now lets set the various headers

             message.Type = RemoveHeader(headers, "type");
             message.Destination = StompHelper.ToDestination(RemoveHeader(headers, "destination"));
             message.ReplyTo = StompHelper.ToDestination(RemoveHeader(headers, "reply-to"));
             message.TargetConsumerId = StompHelper.ToConsumerId(RemoveHeader(headers, "subscription"));
             message.CorrelationId = RemoveHeader(headers, "correlation-id");
             message.MessageId = StompHelper.ToMessageId(RemoveHeader(headers, "message-id"));
             message.Persistent = StompHelper.ToBool(RemoveHeader(headers, "persistent"), true);

             string header = RemoveHeader(headers, "priority");
             if (header != null) message.Priority = Byte.Parse(header);

             header = RemoveHeader(headers, "timestamp");
             if (header != null) message.Timestamp = Int64.Parse(header);

             header = RemoveHeader(headers, "expires");
             if (header != null) message.Expiration = Int64.Parse(header);

             // now lets add the generic headers
             foreach (string key in headers.Keys)
             {
                 Object value = headers[key];
                 if (value != null)
                 {
                     // lets coerce some standard header extensions
                     if (key == "NMSXGroupSeq")
                     {
                         value = Int32.Parse(value.ToString());
                     }
                 }
                 message.Properties[key] = value;
             }
             MessageDispatch dispatch = new MessageDispatch();
             dispatch.Message = message;
             dispatch.ConsumerId = message.TargetConsumerId;
             dispatch.Destination = message.Destination;
             return dispatch;
         }

         protected virtual void WriteConnectionInfo(ConnectionInfo command, StompFrameStream ss)
         {
             // lets force a receipt
             command.ResponseRequired = true;

             ss.WriteCommand(command, "CONNECT");
             ss.WriteHeader("client-id", command.ClientId);
             ss.WriteHeader("login", command.UserName);
             ss.WriteHeader("passcode", command.Password);

             if (command.ResponseRequired)
             {
                 ss.WriteHeader("request-id", command.CommandId);
             }

             ss.Flush();
         }

         protected virtual void WriteShutdownInfo(ShutdownInfo command, StompFrameStream ss)
         {
             ss.WriteCommand(command, "DISCONNECT");
             System.Diagnostics.Debug.Assert(!command.ResponseRequired);
             ss.Flush();
         }

         protected virtual void WriteConsumerInfo(ConsumerInfo command, StompFrameStream ss)
         {
             ss.WriteCommand(command, "SUBSCRIBE");
             ss.WriteHeader("destination", StompHelper.ToStomp(command.Destination));
             ss.WriteHeader("id", StompHelper.ToStomp(command.ConsumerId));
             ss.WriteHeader("durable-subscriber-name", command.SubscriptionName);
             ss.WriteHeader("selector", command.Selector);
             if ( command.NoLocal )
                 ss.WriteHeader("no-local", command.NoLocal);
             ss.WriteHeader("ack", "client");

             // ActiveMQ extensions to STOMP
             ss.WriteHeader("activemq.dispatchAsync", command.DispatchAsync);
             if ( command.Exclusive )
                 ss.WriteHeader("activemq.exclusive", command.Exclusive);

             if( command.SubscriptionName != null )
             {
                 ss.WriteHeader("activemq.subscriptionName", command.SubscriptionName);
                 // For an older 4.0 broker we need to set this header so they get the
                 // subscription as wel..
                 ss.WriteHeader("activemq.subcriptionName", command.SubscriptionName);
             }

             ss.WriteHeader("activemq.maximumPendingMessageLimit", command.MaximumPendingMessageLimit);
             ss.WriteHeader("activemq.prefetchSize", command.PrefetchSize);
             ss.WriteHeader("activemq.priority", command.Priority);
             if ( command.Retroactive )
                 ss.WriteHeader("activemq.retroactive", command.Retroactive);

             consumers[command.ConsumerId] = command.ConsumerId;
             ss.Flush();
         }

         protected virtual void WriteRemoveInfo(RemoveInfo command, StompFrameStream ss)
         {
             object id = command.ObjectId;

             if (id is ConsumerId)
             {
                 ConsumerId consumerId = id as ConsumerId;
                 ss.WriteCommand(command, "UNSUBSCRIBE");
                 ss.WriteHeader("id", StompHelper.ToStomp(consumerId));
                 ss.Flush();
                 consumers.Remove(consumerId);
             }
             else if (id is SessionId)
             {
                 // When a session is removed, it needs to remove it's consumers too.
                 // Find all the consumer that were part of the session.
                 SessionId sessionId = (SessionId) id;
                 ArrayList matches = new ArrayList();
                 foreach (DictionaryEntry entry in consumers)
                 {
                     ConsumerId t = (ConsumerId) entry.Key;
                     if( sessionId.ConnectionId==t.ConnectionId && sessionId.Value==t.SessionId )
                     {
                         matches.Add(t);
                     }
                 }

                 bool unsubscribedConsumer = false;

                 // Un-subscribe them.
                 foreach (ConsumerId consumerId in matches)
                 {
                     ss.WriteCommand(command, "UNSUBSCRIBE");
                     ss.WriteHeader("id", StompHelper.ToStomp(consumerId));
                     ss.Flush();
                     consumers.Remove(consumerId);
                     unsubscribedConsumer = true;
                 }

                 if(!unsubscribedConsumer && command.ResponseRequired)
                 {
                     ss.WriteCommand(command, "UNSUBSCRIBE", true);
                     ss.WriteHeader("id", sessionId);
                     ss.Flush();
                 }
             }
             else if(id is ProducerId)
             {
                 if(command.ResponseRequired)
                 {
                     ss.WriteCommand(command, "UNSUBSCRIBE", true);
                     ss.WriteHeader("id", id);
                     ss.Flush();
                 }
             }
             else if(id is ConnectionId)
             {
                 if(command.ResponseRequired)
                 {
                     ss.WriteCommand(command, "UNSUBSCRIBE", true);
                     ss.WriteHeader("id", id);
                     ss.Flush();
                 }
             }
         }


         protected virtual void WriteTransactionInfo(TransactionInfo command, StompFrameStream ss)
         {
             TransactionId id = command.TransactionId;
             if (id is LocalTransactionId)
             {
                 string type = "BEGIN";
                 TransactionType transactionType = (TransactionType) command.Type;
                 switch (transactionType)
                 {
                     case TransactionType.CommitOnePhase:
                         command.ResponseRequired = true;
                         type = "COMMIT";
                         break;
                     case TransactionType.Rollback:
                         command.ResponseRequired = true;
                         type = "ABORT";
                         break;
                 }

                 Tracer.Debug(">>> For transaction type: " + transactionType + " we are using command type: " + type);
                 ss.WriteCommand(command, type);
                 ss.WriteHeader("transaction", StompHelper.ToStomp(id));
                 ss.Flush();
             }
         }

         protected virtual void WriteMessage(ActiveMQMessage command, StompFrameStream ss)
         {
             ss.WriteCommand(command, "SEND");
             ss.WriteHeader("destination", StompHelper.ToStomp(command.Destination));
             if (command.ReplyTo != null)
                 ss.WriteHeader("reply-to", StompHelper.ToStomp(command.ReplyTo));
             if (command.CorrelationId != null )
                 ss.WriteHeader("correlation-id", command.CorrelationId);
             if (command.Expiration != 0)
                 ss.WriteHeader("expires", command.Expiration);
             if (command.Priority != 4)
                 ss.WriteHeader("priority", command.Priority);
             if (command.Type != null)
                 ss.WriteHeader("type", command.Type);
             if (command.TransactionId!=null)
                 ss.WriteHeader("transaction", StompHelper.ToStomp(command.TransactionId));

             ss.WriteHeader("persistent", command.Persistent);

             // lets force the content to be marshalled

             command.BeforeMarshall(null);
             if (command is ActiveMQTextMessage)
             {
                 ActiveMQTextMessage textMessage = command as ActiveMQTextMessage;
                 ss.Content = encoding.GetBytes(textMessage.Text);
             }
             else
             {
                 ss.Content = command.Content;
                 if(null != command.Content)
                 {
                     ss.ContentLength = command.Content.Length;
                 }
                 else
                 {
                     ss.ContentLength = 0;
                 }
             }

             IPrimitiveMap map = command.Properties;
             foreach (string key in map.Keys)
             {
                 ss.WriteHeader(key, map[key]);
             }
             ss.Flush();
         }

         protected virtual void WriteMessageAck(MessageAck command, StompFrameStream ss)
         {
             ss.WriteCommand(command, "ACK", true);

             // TODO handle bulk ACKs?
             ss.WriteHeader("message-id", StompHelper.ToStomp(command.LastMessageId));
             if(command.TransactionId != null)
             {
                 ss.WriteHeader("transaction", StompHelper.ToStomp(command.TransactionId));
             }

             ss.Flush();
         }

         protected virtual void SendCommand(Command command)
         {
             if (transport == null)
             {
                 Tracer.Fatal("No transport configured so cannot return command: " + command);
             }
             else
             {
                 transport.Command(transport, command);
             }
         }

         protected virtual string RemoveHeader(IDictionary headers, string name)
         {
             object value = headers[name];
             if (value == null)
             {
                 return null;
             }
             else
             {
                 headers.Remove(name);
                 return value.ToString();
             }
         }


         protected virtual string ToString(object value)
         {
             if (value != null)
             {
                 return value.ToString();
             }
             else
             {
                 return null;
             }
         }
     }
 }
	/*
	* 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.
	*/
	using Apache.NMS.ActiveMQ.Commands;
	using Apache.NMS.ActiveMQ.Transport;
	using Apache.NMS;
	using System;
	using System.Collections;
	using System.IO;
	using System.Text;

	namespace Apache.NMS.ActiveMQ.Transport.Stomp
	{
	/// <summary>
	/// Implements the <a href="http://stomp.codehaus.org/">STOMP</a> protocol.
	/// </summary>
	public class StompWireFormat : IWireFormat
	{
	private Encoding encoding = new UTF8Encoding();
	private ITransport transport;
	private IDictionary consumers = Hashtable.Synchronized(new Hashtable());

	public StompWireFormat()
	{
	}

	public ITransport Transport {
	get { return transport; }
	set { transport = value; }
	}

	public int Version {
	get { return 1; }
	}

	public void Marshal(Object o, BinaryWriter binaryWriter)
	{
	Tracer.Debug(">>>> " + o);
	StompFrameStream ds = new StompFrameStream(binaryWriter, encoding);

	if (o is ConnectionInfo)
	{
	WriteConnectionInfo((ConnectionInfo) o, ds);
	}
	else if (o is ActiveMQMessage)
	{
	WriteMessage((ActiveMQMessage) o, ds);
	}
	else if (o is ConsumerInfo)
	{
	WriteConsumerInfo((ConsumerInfo) o, ds);
	}
	else if (o is MessageAck)
	{
	WriteMessageAck((MessageAck) o, ds);
	}
	else if (o is TransactionInfo)
	{
	WriteTransactionInfo((TransactionInfo) o, ds);
	}
	else if (o is ShutdownInfo)
	{
	WriteShutdownInfo((ShutdownInfo) o, ds);
	}
	else if (o is RemoveInfo)
	{
	WriteRemoveInfo((RemoveInfo) o, ds);
	}
	else if (o is Command)
	{
	Command command = o as Command;
	if (command.ResponseRequired)
	{
	Response response = new Response();
	response.CorrelationId = command.CommandId;
	SendCommand(response);
	Tracer.Debug("#### Autorespond to command: " + o.GetType());
	}
	}
	else
	{
	Tracer.Debug("#### Ignored command: " + o.GetType());
	}
	}


	internal String ReadLine(BinaryReader dis)
	{
	MemoryStream ms = new MemoryStream();
	while (true)
	{
	int nextChar = dis.Read();
	if (nextChar < 0)
	{
	throw new IOException("Peer closed the stream.");
	}
	if( nextChar == 10 )
	{
	break;
	}
	ms.WriteByte((byte)nextChar);
	}
	byte[] data = ms.ToArray();
	return encoding.GetString(data, 0, data.Length);
	}

	public Object Unmarshal(BinaryReader dis)
	{
	string command;
	do {
	command = ReadLine(dis);
	}
	while (command == "");

	Tracer.Debug("<<<< command: " + command);

	IDictionary headers = new Hashtable();
	string line;
	while ((line = ReadLine(dis)) != "")
	{
	int idx = line.IndexOf(':');
	if (idx > 0)
	{
	string key = line.Substring(0, idx);
	string value = line.Substring(idx + 1);
	headers[key] = value;

	Tracer.Debug("<<<< header: " + key + " = " + value);
	}
	else
	{
	// lets ignore this bad header!
	}
	}
	byte[] content = null;
	string length = ToString(headers["content-length"]);
	if (length != null)
	{
	int size = Int32.Parse(length);
	content = dis.ReadBytes(size);
	// Read the terminating NULL byte for this frame.
	int nullByte = dis.Read();
	if(nullByte != 0)
	{
	Tracer.Debug("<<<< error reading frame null byte.");
	}
	}
	else
	{
	MemoryStream ms = new MemoryStream();
	int nextChar;
	while((nextChar = dis.Read()) != 0)
	{
	if( nextChar < 0 )
	{
	// EOF ??
	break;
	}
	ms.WriteByte((byte)nextChar);
	}
	content = ms.ToArray();
	}
	Object answer = CreateCommand(command, headers, content);
	Tracer.Debug("<<<< received: " + answer);
	return answer;
	}

	protected virtual Object CreateCommand(string command, IDictionary headers, byte[] content)
	{
	if(command == "RECEIPT" \|\| command == "CONNECTED")
	{
	string text = RemoveHeader(headers, "receipt-id");
	if(text != null)
	{
	Response answer = new Response();
	if(text.StartsWith("ignore:"))
	{
	text = text.Substring("ignore:".Length);
	}

	answer.CorrelationId = Int32.Parse(text);
	return answer;
	}
	else if(command == "CONNECTED")
	{
	text = RemoveHeader(headers, "response-id");
	if (text != null)
	{
	Response answer = new Response();
	answer.CorrelationId = Int32.Parse(text);
	return answer;
	}
	}
	}
	else if(command == "ERROR")
	{
	string text = RemoveHeader(headers, "receipt-id");

	if(text != null && text.StartsWith("ignore:"))
	{
	Response answer = new Response();
	answer.CorrelationId = Int32.Parse(text.Substring("ignore:".Length));
	return answer;
	}
	else
	{
	ExceptionResponse answer = new ExceptionResponse();
	if(text != null)
	{
	answer.CorrelationId = Int32.Parse(text);
	}

	BrokerError error = new BrokerError();
	error.Message = RemoveHeader(headers, "message");
	error.ExceptionClass = RemoveHeader(headers, "exceptionClass");
	// TODO is this the right header?
	answer.Exception = error;
	return answer;
	}
	}
	else if (command == "MESSAGE")
	{
	return ReadMessage(command, headers, content);
	}
	Tracer.Error("Unknown command: " + command + " headers: " + headers);
	return null;
	}

	protected virtual Command ReadMessage(string command, IDictionary headers, byte[] content)
	{
	ActiveMQMessage message = null;
	if (headers.Contains("content-length"))
	{
	message = new ActiveMQBytesMessage();
	message.Content = content;
	}
	else
	{
	message = new ActiveMQTextMessage(encoding.GetString(content, 0, content.Length));
	}

	// TODO now lets set the various headers

	message.Type = RemoveHeader(headers, "type");
	message.Destination = StompHelper.ToDestination(RemoveHeader(headers, "destination"));
	message.ReplyTo = StompHelper.ToDestination(RemoveHeader(headers, "reply-to"));
	message.TargetConsumerId = StompHelper.ToConsumerId(RemoveHeader(headers, "subscription"));
	message.CorrelationId = RemoveHeader(headers, "correlation-id");
	message.MessageId = StompHelper.ToMessageId(RemoveHeader(headers, "message-id"));
	message.Persistent = StompHelper.ToBool(RemoveHeader(headers, "persistent"), true);

	string header = RemoveHeader(headers, "priority");
	if (header != null) message.Priority = Byte.Parse(header);

	header = RemoveHeader(headers, "timestamp");
	if (header != null) message.Timestamp = Int64.Parse(header);

	header = RemoveHeader(headers, "expires");
	if (header != null) message.Expiration = Int64.Parse(header);

	// now lets add the generic headers
	foreach (string key in headers.Keys)
	{
	Object value = headers[key];
	if (value != null)
	{
	// lets coerce some standard header extensions
	if (key == "NMSXGroupSeq")
	{
	value = Int32.Parse(value.ToString());
	}
	}
	message.Properties[key] = value;
	}
	MessageDispatch dispatch = new MessageDispatch();
	dispatch.Message = message;
	dispatch.ConsumerId = message.TargetConsumerId;
	dispatch.Destination = message.Destination;
	return dispatch;
	}

	protected virtual void WriteConnectionInfo(ConnectionInfo command, StompFrameStream ss)
	{
	// lets force a receipt
	command.ResponseRequired = true;

	ss.WriteCommand(command, "CONNECT");
	ss.WriteHeader("client-id", command.ClientId);
	ss.WriteHeader("login", command.UserName);
	ss.WriteHeader("passcode", command.Password);

	if (command.ResponseRequired)
	{
	ss.WriteHeader("request-id", command.CommandId);
	}

	ss.Flush();
	}

	protected virtual void WriteShutdownInfo(ShutdownInfo command, StompFrameStream ss)
	{
	ss.WriteCommand(command, "DISCONNECT");
	System.Diagnostics.Debug.Assert(!command.ResponseRequired);
	ss.Flush();
	}

	protected virtual void WriteConsumerInfo(ConsumerInfo command, StompFrameStream ss)
	{
	ss.WriteCommand(command, "SUBSCRIBE");
	ss.WriteHeader("destination", StompHelper.ToStomp(command.Destination));
	ss.WriteHeader("id", StompHelper.ToStomp(command.ConsumerId));
	ss.WriteHeader("durable-subscriber-name", command.SubscriptionName);
	ss.WriteHeader("selector", command.Selector);
	if ( command.NoLocal )
	ss.WriteHeader("no-local", command.NoLocal);
	ss.WriteHeader("ack", "client");

	// ActiveMQ extensions to STOMP
	ss.WriteHeader("activemq.dispatchAsync", command.DispatchAsync);
	if ( command.Exclusive )
	ss.WriteHeader("activemq.exclusive", command.Exclusive);

	if( command.SubscriptionName != null )
	{
	ss.WriteHeader("activemq.subscriptionName", command.SubscriptionName);
	// For an older 4.0 broker we need to set this header so they get the
	// subscription as wel..
	ss.WriteHeader("activemq.subcriptionName", command.SubscriptionName);
	}

	ss.WriteHeader("activemq.maximumPendingMessageLimit", command.MaximumPendingMessageLimit);
	ss.WriteHeader("activemq.prefetchSize", command.PrefetchSize);
	ss.WriteHeader("activemq.priority", command.Priority);
	if ( command.Retroactive )
	ss.WriteHeader("activemq.retroactive", command.Retroactive);

	consumers[command.ConsumerId] = command.ConsumerId;
	ss.Flush();
	}

	protected virtual void WriteRemoveInfo(RemoveInfo command, StompFrameStream ss)
	{
	object id = command.ObjectId;

	if (id is ConsumerId)
	{
	ConsumerId consumerId = id as ConsumerId;
	ss.WriteCommand(command, "UNSUBSCRIBE");
	ss.WriteHeader("id", StompHelper.ToStomp(consumerId));
	ss.Flush();
	consumers.Remove(consumerId);
	}
	else if (id is SessionId)
	{
	// When a session is removed, it needs to remove it's consumers too.
	// Find all the consumer that were part of the session.
	SessionId sessionId = (SessionId) id;
	ArrayList matches = new ArrayList();
	foreach (DictionaryEntry entry in consumers)
	{
	ConsumerId t = (ConsumerId) entry.Key;
	if( sessionId.ConnectionId==t.ConnectionId && sessionId.Value==t.SessionId )
	{
	matches.Add(t);
	}
	}

	bool unsubscribedConsumer = false;

	// Un-subscribe them.
	foreach (ConsumerId consumerId in matches)
	{
	ss.WriteCommand(command, "UNSUBSCRIBE");
	ss.WriteHeader("id", StompHelper.ToStomp(consumerId));
	ss.Flush();
	consumers.Remove(consumerId);
	unsubscribedConsumer = true;
	}

	if(!unsubscribedConsumer && command.ResponseRequired)
	{
	ss.WriteCommand(command, "UNSUBSCRIBE", true);
	ss.WriteHeader("id", sessionId);
	ss.Flush();
	}
	}
	else if(id is ProducerId)
	{
	if(command.ResponseRequired)
	{
	ss.WriteCommand(command, "UNSUBSCRIBE", true);
	ss.WriteHeader("id", id);
	ss.Flush();
	}
	}
	else if(id is ConnectionId)
	{
	if(command.ResponseRequired)
	{
	ss.WriteCommand(command, "UNSUBSCRIBE", true);
	ss.WriteHeader("id", id);
	ss.Flush();
	}
	}
	}


	protected virtual void WriteTransactionInfo(TransactionInfo command, StompFrameStream ss)
	{
	TransactionId id = command.TransactionId;
	if (id is LocalTransactionId)
	{
	string type = "BEGIN";
	TransactionType transactionType = (TransactionType) command.Type;
	switch (transactionType)
	{
	case TransactionType.CommitOnePhase:
	command.ResponseRequired = true;
	type = "COMMIT";
	break;
	case TransactionType.Rollback:
	command.ResponseRequired = true;
	type = "ABORT";
	break;
	}

	Tracer.Debug(">>> For transaction type: " + transactionType + " we are using command type: " + type);
	ss.WriteCommand(command, type);
	ss.WriteHeader("transaction", StompHelper.ToStomp(id));
	ss.Flush();
	}
	}

	protected virtual void WriteMessage(ActiveMQMessage command, StompFrameStream ss)
	{
	ss.WriteCommand(command, "SEND");
	ss.WriteHeader("destination", StompHelper.ToStomp(command.Destination));
	if (command.ReplyTo != null)
	ss.WriteHeader("reply-to", StompHelper.ToStomp(command.ReplyTo));
	if (command.CorrelationId != null )
	ss.WriteHeader("correlation-id", command.CorrelationId);
	if (command.Expiration != 0)
	ss.WriteHeader("expires", command.Expiration);
	if (command.Priority != 4)
	ss.WriteHeader("priority", command.Priority);
	if (command.Type != null)
	ss.WriteHeader("type", command.Type);
	if (command.TransactionId!=null)
	ss.WriteHeader("transaction", StompHelper.ToStomp(command.TransactionId));

	ss.WriteHeader("persistent", command.Persistent);

	// lets force the content to be marshalled

	command.BeforeMarshall(null);
	if (command is ActiveMQTextMessage)
	{
	ActiveMQTextMessage textMessage = command as ActiveMQTextMessage;
	ss.Content = encoding.GetBytes(textMessage.Text);
	}
	else
	{
	ss.Content = command.Content;
	if(null != command.Content)
	{
	ss.ContentLength = command.Content.Length;
	}
	else
	{
	ss.ContentLength = 0;
	}
	}

	IPrimitiveMap map = command.Properties;
	foreach (string key in map.Keys)
	{
	ss.WriteHeader(key, map[key]);
	}
	ss.Flush();
	}

	protected virtual void WriteMessageAck(MessageAck command, StompFrameStream ss)
	{
	ss.WriteCommand(command, "ACK", true);

	// TODO handle bulk ACKs?
	ss.WriteHeader("message-id", StompHelper.ToStomp(command.LastMessageId));
	if(command.TransactionId != null)
	{
	ss.WriteHeader("transaction", StompHelper.ToStomp(command.TransactionId));
	}

	ss.Flush();
	}

	protected virtual void SendCommand(Command command)
	{
	if (transport == null)
	{
	Tracer.Fatal("No transport configured so cannot return command: " + command);
	}
	else
	{
	transport.Command(transport, command);
	}
	}

	protected virtual string RemoveHeader(IDictionary headers, string name)
	{
	object value = headers[name];
	if (value == null)
	{
	return null;
	}
	else
	{
	headers.Remove(name);
	return value.ToString();
	}
	}


	protected virtual string ToString(object value)
	{
	if (value != null)
	{
	return value.ToString();
	}
	else
	{
	return null;
	}
	}
	}
	}