Author: tabish
Date: Sat Nov 16 19:08:02 2013
New Revision: 1542552
URL: http://svn.apache.org/r1542552
Log:
https://issues.apache.org/jira/browse/AMQNET-458
Implementation
Added:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBCOMP.cs
(with props)
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs
(with props)
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNACK.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNECT.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/DISCONNECT.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGREQ.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGRESP.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBACK.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBLISH.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREC.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREL.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBACK.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBSCRIBE.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBACK.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBSCRIBE.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Connection.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageConsumer.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageProducer.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MQTTMessage.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MessageDispatch.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Session.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/BaseCommand.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Command.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Response.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/TransportFactory.cs
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/vs2008-mqtt.csproj
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNACK.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNACK.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNACK.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNACK.cs
Sat Nov 16 19:08:02 2013
@@ -15,10 +15,11 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class CONNACK
+ public class CONNACK : BaseCommand
{
public const byte TYPE = 2;
@@ -29,20 +30,20 @@ namespace Apache.NMS.MQTT.Commands
set { this.returnCode = value; }
}
- public CONNACK()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "CONNACK"; }
}
+ public override bool IsCONNACK
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNECT.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNECT.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNECT.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/CONNECT.cs
Sat Nov 16 19:08:02 2013
@@ -15,6 +15,7 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
@@ -26,7 +27,7 @@ namespace Apache.NMS.MQTT.Commands
/// to use. All but the first are optional and their presence is
determined based on flags
/// in the variable header.
/// </summary>
- public class CONNECT
+ public class CONNECT : BaseCommand
{
public const byte TYPE = 1;
public const String PROTOCOL_NAME = "MQIsdp";
@@ -36,11 +37,16 @@ namespace Apache.NMS.MQTT.Commands
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "CONNECT"; }
}
+ public override bool IsCONNECT
+ {
+ get { return true; }
+ }
+
private byte version = 3;
public byte Version
{
@@ -91,7 +97,7 @@ namespace Apache.NMS.MQTT.Commands
}
private short keepAliveTimer = 10;
- public bool KeepAliveTimer
+ public short KeepAliveTimer
{
get { return this.keepAliveTimer; }
set { this.keepAliveTimer = value; }
@@ -110,10 +116,6 @@ namespace Apache.NMS.MQTT.Commands
get { return this.willMessage; }
set { this.willMessage = value; }
}
-
- public CONNECT()
- {
- }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/DISCONNECT.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/DISCONNECT.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/DISCONNECT.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/DISCONNECT.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class DISCONNECT
+ public class DISCONNECT : BaseCommand
{
public const byte TYPE = 14;
- public DISCONNECT()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "DISCONNECT"; }
}
+
+ public override bool IsDISCONNECT
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGREQ.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGREQ.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGREQ.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGREQ.cs
Sat Nov 16 19:08:02 2013
@@ -15,27 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class PINGREQ
+ public class PINGREQ : BaseCommand
{
public const byte TYPE = 11;
- public PINGREQ()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "PINGREQ"; }
}
+ public override bool IsPINGREQ
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGRESP.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGRESP.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGRESP.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PINGRESP.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class PINGRESP
+ public class PINGRESP : BaseCommand
{
public const byte TYPE = 13;
- public PINGRESP()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "PINGRESP"; }
}
+
+ public override bool IsPINGRESP
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBACK.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBACK.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBACK.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBACK.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class PUBACK
+ public class PUBACK : BaseCommand
{
public const byte TYPE = 4;
- public PUBACK()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "PUBACK"; }
}
+
+ public override bool IsPUBACK
+ {
+ get { return true; }
+ }
}
}
Added:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBCOMP.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBCOMP.cs?rev=1542552&view=auto
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBCOMP.cs
(added)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBCOMP.cs
Sat Nov 16 19:08:02 2013
@@ -0,0 +1,42 @@
+//
+// 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 System;
+using Apache.NMS.MQTT.Transport;
+
+namespace Apache.NMS.MQTT.Commands
+{
+ public class PUBCOMP : BaseCommand
+ {
+ public const byte TYPE = 7;
+
+ public int CommandType
+ {
+ get { return TYPE; }
+ }
+
+ public string CommandName
+ {
+ get { return "PUBCOMP"; }
+ }
+
+ public override bool IsPUBCOMP
+ {
+ get { return true; }
+ }
+ }
+}
+
Propchange:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBCOMP.cs
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBLISH.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBLISH.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBLISH.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBLISH.cs
Sat Nov 16 19:08:02 2013
@@ -15,6 +15,7 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
@@ -27,7 +28,7 @@ namespace Apache.NMS.MQTT.Commands
/// in the application the appropriate payload flag fields to handle the
compression details.
/// You cannot define application-specific flags in the fixed or variable
headers.
/// </summary>
- public class PUBLISH
+ public class PUBLISH : BaseCommand
{
public const byte TYPE = 3;
@@ -36,11 +37,16 @@ namespace Apache.NMS.MQTT.Commands
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "PUBLISH"; }
}
+ public override bool IsPUBLISH
+ {
+ get { return true; }
+ }
+
private byte qosLevel;
public byte QoSLevel
{
@@ -82,10 +88,6 @@ namespace Apache.NMS.MQTT.Commands
get { return this.payload; }
set { this.payload = value; }
}
-
- public PUBLISH()
- {
- }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREC.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREC.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREC.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREC.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class PUBREC
+ public class PUBREC : BaseCommand
{
public const byte TYPE = 5;
- public PUBREC()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "PUBREC"; }
}
+
+ public override bool IsPUBREC
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREL.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREL.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREL.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/PUBREL.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class PUBREL
+ public class PUBREL : BaseCommand
{
public const byte TYPE = 6;
- public PUBREL()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "PUBREL"; }
}
+
+ public override bool IsPUBREL
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBACK.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBACK.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBACK.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBACK.cs
Sat Nov 16 19:08:02 2013
@@ -15,6 +15,7 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
@@ -24,23 +25,24 @@ namespace Apache.NMS.MQTT.Commands
/// particular Topic Name. Granted QoS levels are listed in the same order
as the topic
/// names in the corresponding SUBSCRIBE message.
/// </summary>
- public class SUBACK
+ public class SUBACK : BaseCommand
{
public const byte TYPE = 8;
- public SUBACK()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "SUBACK"; }
}
+
+ public override bool IsSUBACK
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBSCRIBE.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBSCRIBE.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBSCRIBE.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/SUBSCRIBE.cs
Sat Nov 16 19:08:02 2013
@@ -15,6 +15,7 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
@@ -22,23 +23,24 @@ namespace Apache.NMS.MQTT.Commands
/// The payload contains a list of topic names to which the client can
subscribe, and
/// the QoS level. These strings are UTF-encoded.
/// </summary>
- public class SUBSCRIBE
+ public class SUBSCRIBE : BaseCommand
{
public const byte TYPE = 7;
- public SUBSCRIBE()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "SUBSCRIBE"; }
}
+
+ public override bool IsSUBSCRIBE
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBACK.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBACK.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBACK.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBACK.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,28 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class UNSUBACK
+ public class UNSUBACK : BaseCommand
{
public const byte TYPE = 10;
- public UNSUBACK()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "UNSUBACK"; }
}
+
+ public override bool IsUNSUBACK
+ {
+ get { return true; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBSCRIBE.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBSCRIBE.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBSCRIBE.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Commands/UNSUBSCRIBE.cs
Sat Nov 16 19:08:02 2013
@@ -15,26 +15,35 @@
// limitations under the License.
//
using System;
+using Apache.NMS.MQTT.Transport;
namespace Apache.NMS.MQTT.Commands
{
- public class UNSUBSCRIBE
+ public class UNSUBSCRIBE : BaseCommand
{
public const byte TYPE = 9;
- public UNSUBSCRIBE()
- {
- }
-
public int CommandType
{
get { return TYPE; }
}
- public int CommandName
+ public string CommandName
{
get { return "UNSUBSCRIBE"; }
}
+
+ public override bool IsUNSUBSCRIBE
+ {
+ get { return true; }
+ }
+
+ private string[] topics;
+ public string[] Topics
+ {
+ get { return this.topics; }
+ set { this.topics = value; }
+ }
}
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Connection.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Connection.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Connection.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Connection.cs
Sat Nov 16 19:08:02 2013
@@ -53,6 +53,7 @@ namespace Apache.NMS.MQTT
private readonly MessageTransformation messageTransformation;
private readonly ThreadPoolExecutor executor = new
ThreadPoolExecutor();
private readonly IdGenerator clientIdGenerator;
+ private IRedeliveryPolicy redeliveryPolicy;
public Connection(Uri connectionUri, ITransport transport,
IdGenerator clientIdGenerator)
{
@@ -195,6 +196,15 @@ namespace Apache.NMS.MQTT
get { return this.messageTransformation; }
}
+ /// <summary>
+ /// Get/or set the redelivery policy for this connection.
+ /// </summary>
+ public IRedeliveryPolicy RedeliveryPolicy
+ {
+ get { return this.redeliveryPolicy; }
+ set { this.redeliveryPolicy = value; }
+ }
+
#endregion
private void SetTransport(ITransport newTransport)
@@ -357,6 +367,11 @@ namespace Apache.NMS.MQTT
disposed = true;
}
+ public void PurgeTempDestinations()
+ {
+ throw new NotSupportedException("MQTT does not support
temp destinations.");
+ }
+
protected void OnCommand(ITransport commandTransport, Command
command)
{
}
@@ -435,35 +450,35 @@ namespace Apache.NMS.MQTT
// Send the connection and see if an ack/nak is returned.
Response response = transport.Request(this.info, this.RequestTimeout);
-
if(!(response is ExceptionResponse))
+
if(!response.IsErrorResponse)
{
connected.Value = true;
}
else
{
-
ExceptionResponse error = response as ExceptionResponse;
-
NMSException exception =
CreateExceptionFromBrokerError(error.Exception);
-
if(exception is InvalidClientIDException)
-
{
-
// This is non-recoverable.
-
// Shutdown the transport connection, and re-create it, but
don't start it.
-
// It will be started if the connection is re-attempted.
-
this.transport.Stop();
-
ITransport newTransport =
TransportFactory.CreateTransport(this.brokerUri);
-
SetTransport(newTransport);
-
throw exception;
-
}
+
// TODO figure out correct error to throw.
+//
NMSException exception =
CreateExceptionFromBrokerError(error.Exception);
+//
if(exception is InvalidClientIDException)
+//
{
+//
// This is non-recoverable.
+//
// Shutdown the transport connection, and re-create it, but
don't start it.
+//
// It will be started if the connection is re-attempted.
+//
this.transport.Stop();
+//
ITransport newTransport =
TransportFactory.CreateTransport(this.brokerUri);
+//
SetTransport(newTransport);
+//
throw exception;
+//
}
}
}
}
-
catch(BrokerException)
- {
- // We
Swallow the generic version and throw ConnectionClosedException
- }
catch(NMSException)
{
throw;
}
+ catch(Exception)
+ {
+ // We
Swallow the generic version and throw ConnectionClosedException
+ }
}
}
finally
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageConsumer.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageConsumer.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageConsumer.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageConsumer.cs
Sat Nov 16 19:08:02 2013
@@ -36,6 +36,7 @@ namespace Apache.NMS.MQTT
private Exception failureError;
private ThreadPoolExecutor executor;
+ private int consumerId;
private event MessageListener listener;
@@ -77,7 +78,7 @@ namespace Apache.NMS.MQTT
}
listener += value;
-
this.session.Redispatch(this.unconsumedMessages);
+
//this.session.Redispatch(this.unconsumedMessages);
if(wasStarted)
{
@@ -87,6 +88,11 @@ namespace Apache.NMS.MQTT
remove { listener -= value; }
}
+ public int ConsumerId
+ {
+ get { return this.consumerId; }
+ }
+
#endregion
public void Start()
@@ -114,7 +120,7 @@ namespace Apache.NMS.MQTT
MessageDispatch dispatch =
this.unconsumedMessages.DequeueNoWait();
if(dispatch != null)
{
- this.Dispatch(dispatch);
+ //this.Dispatch(dispatch);
return true;
}
}
@@ -138,18 +144,9 @@ namespace Apache.NMS.MQTT
}
}
- protected bool IsAutoAcknowledgeEach
- {
- get
- {
- return this.session.IsAutoAcknowledge ||
- (this.session.IsDupsOkAcknowledge &&
this.info.Destination.IsQueue);
- }
- }
-
- protected bool IsAutoAcknowledgeBatch
+ protected bool IsAutoAcknowledge
{
- get { return this.session.IsDupsOkAcknowledge &&
!this.info.Destination.IsQueue; }
+ get { return this.session.IsAutoAcknowledge; }
}
protected bool IsIndividualAcknowledge
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageProducer.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageProducer.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageProducer.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/MessageProducer.cs
Sat Nov 16 19:08:02 2013
@@ -31,6 +31,7 @@ namespace Apache.NMS.MQTT
private MsgDeliveryMode msgDeliveryMode =
NMSConstants.defaultDeliveryMode;
private TimeSpan requestTimeout;
protected bool disposed = false;
+ private int producerId;
private readonly MessageTransformation messageTransformation;
@@ -46,6 +47,21 @@ namespace Apache.NMS.MQTT
Dispose(false);
}
+ #region MessageProducer Property Accessors
+
+ public int ProducerId
+ {
+ get { return this.producerId; }
+ }
+
+ public TimeSpan RequestTimeout
+ {
+ get { return requestTimeout; }
+ set { this.requestTimeout = value; }
+ }
+
+ #endregion
+
public void Dispose()
{
Dispose(true);
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MQTTMessage.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MQTTMessage.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MQTTMessage.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MQTTMessage.cs
Sat Nov 16 19:08:02 2013
@@ -23,12 +23,13 @@ namespace Apache.NMS.MQTT.Messages
{
public delegate void AcknowledgeHandler(MQTTMessage message);
- public class MQTTMessage : IMessage
+ public class MQTTMessage : IMessage, ICloneable
{
private readonly PUBLISH publish = new PUBLISH();
private MessagePropertyIntercepter propertyHelper;
private PrimitiveMap properties;
private Connection connection;
+ private Topic destination;
public event AcknowledgeHandler Acknowledger;
@@ -39,7 +40,6 @@ namespace Apache.NMS.MQTT.Messages
public MQTTMessage() : base()
{
- Timestamp = DateUtils.ToJavaTimeUtc(DateTime.UtcNow);
}
public override int GetHashCode()
@@ -133,12 +133,6 @@ namespace Apache.NMS.MQTT.Messages
}
}
- public IDestination FromDestination
- {
- get { return Destination; }
- set { this.Destination =
ActiveMQDestination.Transform(value); }
- }
-
public Connection Connection
{
get { return this.connection; }
@@ -301,59 +295,8 @@ namespace Apache.NMS.MQTT.Messages
/// </summary>
public string NMSType
{
- get { return Type; }
- set { Type = value; }
- }
-
- #endregion
-
- #region NMS Extension headers
-
- /// <summary>
- /// Returns the number of times this message has been
redelivered to other consumers without being acknowledged successfully.
- /// </summary>
- public int NMSXDeliveryCount
- {
- get { return RedeliveryCounter + 1; }
- }
-
- /// <summary>
- /// The Message Group ID used to group messages together to the
same consumer for the same group ID value
- /// </summary>
- public string NMSXGroupID
- {
- get { return GroupID; }
- set { GroupID = value; }
- }
- /// <summary>
- /// The Message Group Sequence counter to indicate the position
in a group
- /// </summary>
- public int NMSXGroupSeq
- {
- get { return GroupSequence; }
- set { GroupSequence = value; }
- }
-
- /// <summary>
- /// Returns the ID of the producers transaction
- /// </summary>
- public string NMSXProducerTXID
- {
- get
- {
- TransactionId txnId = OriginalTransactionId;
- if(null == txnId)
- {
- txnId = TransactionId;
- }
-
- if(null != txnId)
- {
- return
BaseDataStreamMarshaller.ToString(txnId);
- }
-
- return String.Empty;
- }
+ get { return publish.CommandName; }
+ set { }
}
#endregion
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MessageDispatch.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MessageDispatch.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MessageDispatch.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Messages/MessageDispatch.cs
Sat Nov 16 19:08:02 2013
@@ -49,12 +49,6 @@ namespace Apache.NMS.MQTT.Messages
"Message = " + Message + " ]";
}
- public Exception RollbackCause
- {
- get { return this.rollbackCause; }
- set { this.rollbackCause = value; }
- }
-
public Topic Destination
{
get { return destination; }
Added:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs?rev=1542552&view=auto
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs
(added)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs
Sat Nov 16 19:08:02 2013
@@ -0,0 +1,52 @@
+/*
+ * 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 System;
+using System.Text;
+
+using Apache.NMS;
+
+namespace Apache.NMS.MQTT
+{
+ /// <summary>
+ /// Exception thrown when an Request times out.
+ /// </summary>
+ public class RequestTimedOutException : IOException
+ {
+ public RequestTimedOutException()
+ : base("Synchronous Request Timed out")
+ {
+ }
+
+ public RequestTimedOutException(TimeSpan interval)
+ : base("Synchronous Request Timed out after [" +
interval.TotalMilliseconds + "] milliseconds")
+ {
+ }
+
+ public RequestTimedOutException(String msg)
+ : base(msg)
+ {
+ }
+
+ public RequestTimedOutException(String msg, Exception inner)
+ : base(msg, inner)
+ {
+ }
+ }
+}
+
+
Propchange:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/RequestTimedOutException.cs
------------------------------------------------------------------------------
svn:executable = *
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Session.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Session.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Session.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Session.cs
Sat Nov 16 19:08:02 2013
@@ -39,8 +39,11 @@ namespace Apache.NMS.MQTT
protected bool disposed = false;
protected bool closed = false;
protected bool closing = false;
+ private int sessionId;
private readonly AcknowledgementMode acknowledgementMode;
+ private TimeSpan disposeStopTimeout = TimeSpan.FromMilliseconds(30000);
+ private TimeSpan requestTimeout;
public Session(Connection connection, AcknowledgementMode
acknowledgementMode)
{
@@ -65,6 +68,31 @@ namespace Apache.NMS.MQTT
Dispose(false);
}
+ #region Session Transaction Events
+
+ // We delegate the events to the TransactionContext since it knows
+ // what the state is for both Local and DTC transactions.
+
+ public event SessionTxEventDelegate TransactionStartedListener
+ {
+ add { throw new NotSupportedException("MQTT Does not
implement transactions"); }
+ remove { throw new NotSupportedException("MQTT Does not implement
transactions"); }
+ }
+
+ public event SessionTxEventDelegate TransactionCommittedListener
+ {
+ add { throw new NotSupportedException("MQTT Does not implement
transactions"); }
+ remove { throw new NotSupportedException("MQTT Does not implement
transactions"); }
+ }
+
+ public event SessionTxEventDelegate TransactionRolledBackListener
+ {
+ add { throw new NotSupportedException("MQTT Does not implement
transactions"); }
+ remove { throw new NotSupportedException("MQTT Does not implement
transactions"); }
+ }
+
+ #endregion
+
#region Property Accessors
public virtual AcknowledgementMode AcknowledgementMode
@@ -97,11 +125,27 @@ namespace Apache.NMS.MQTT
get{ return this.acknowledgementMode ==
AcknowledgementMode.Transactional; }
}
+ public Connection Connection
+ {
+ get { return this.connection; }
+ }
+
+ public bool Transacted
+ {
+ get { return this.IsTransacted; }
+ }
+
public SessionExecutor Executor
{
get { return this.executor; }
}
+ public TimeSpan RequestTimeout
+ {
+ get { return this.requestTimeout; }
+ set { this.requestTimeout = value; }
+ }
+
private ConsumerTransformerDelegate consumerTransformer;
/// <summary>
/// A Delegate that is called each time a Message is dispatched to
allow the client to do
@@ -164,9 +208,9 @@ namespace Apache.NMS.MQTT
{
try
{
- Tracer.InfoFormat("Closing The Session with Id {0}",
this.info.SessionId);
+ Tracer.InfoFormat("Closing The Session with Id {0}",
this.sessionId);
DoClose();
- Tracer.InfoFormat("Closed The Session with Id {0}",
this.info.SessionId);
+ Tracer.InfoFormat("Closed The Session with Id {0}",
this.sessionId);
}
catch (Exception ex)
{
@@ -175,6 +219,10 @@ namespace Apache.NMS.MQTT
}
}
+ internal void DoClose()
+ {
+ }
+
public IMessageProducer CreateProducer()
{
return CreateProducer(null);
@@ -249,6 +297,11 @@ namespace Apache.NMS.MQTT
throw new NotSupportedException("Not supported with MQTT
Protocol");
}
+ public void DeleteDurableConsumer(string name)
+ {
+ throw new NotSupportedException("MQTT Cannot delete Durable
Consumers");
+ }
+
/// <summary>
/// Delete a destination (Queue, Topic, Temp Queue, Temp Topic).
/// </summary>
@@ -356,17 +409,17 @@ namespace Apache.NMS.MQTT
{
if(!this.closing)
{
- ConsumerId id = consumer.ConsumerId;
+ int id = consumer.ConsumerId;
// Registered with Connection before we register at the broker.
consumers[id] = consumer;
- connection.AddDispatcher(id, this);
+ //connection.AddDispatcher(id, this);
}
}
public void RemoveConsumer(MessageConsumer consumer)
{
- connection.RemoveDispatcher(consumer.ConsumerId);
+ //connection.RemoveDispatcher(consumer.ConsumerId);
if(!this.closing)
{
consumers.Remove(consumer.ConsumerId);
@@ -378,10 +431,10 @@ namespace Apache.NMS.MQTT
{
if(!this.closing)
{
- ProducerId id = producer.ProducerId;
+ int id = producer.ProducerId;
this.producers[id] = producer;
- this.connection.AddProducer(id, producer);
+ //this.connection.AddProducer(id, producer);
}
}
@@ -394,6 +447,14 @@ namespace Apache.NMS.MQTT
// }
// }
+ public void Dispatch(MessageDispatch dispatch)
+ {
+ if(this.executor != null)
+ {
+ this.executor.Execute(dispatch);
+ }
+ }
+
private MQTTMessage ConfigureMessage(MQTTMessage message)
{
message.Connection = this.connection;
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/BaseCommand.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/BaseCommand.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/BaseCommand.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/BaseCommand.cs
Sat Nov 16 19:08:02 2013
@@ -116,6 +116,11 @@ namespace Apache.NMS.MQTT.Transport
get { return false; }
}
+ public virtual bool IsPUBCOMP
+ {
+ get { return false; }
+ }
+
public virtual bool IsSUBACK
{
get { return false; }
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Command.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Command.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Command.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Command.cs
Sat Nov 16 19:08:02 2013
@@ -99,6 +99,11 @@ namespace Apache.NMS.MQTT.Transport
get;
}
+ bool IsPUBCOMP
+ {
+ get;
+ }
+
bool IsSUBACK
{
get;
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Response.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Response.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Response.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/Response.cs
Sat Nov 16 19:08:02 2013
@@ -55,6 +55,11 @@ namespace Apache.NMS.MQTT.Transport
get { return true; }
}
+ public bool IsErrorResponse
+ {
+ get { return false; }
+ }
+
};
}
Modified:
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/TransportFactory.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/TransportFactory.cs?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/TransportFactory.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/src/main/csharp/Transport/TransportFactory.cs
Sat Nov 16 19:08:02 2013
@@ -27,7 +27,7 @@ namespace Apache.NMS.MQTT.Transport
public static event ExceptionListener OnException;
private static readonly FactoryFinder<MQTTTransportFactoryAttribute,
ITransportFactory> FACTORY_FINDER =
- new FactoryFinder<ActiveMQTransportFactoryAttribute,
ITransportFactory>();
+ new FactoryFinder<MQTTTransportFactoryAttribute,
ITransportFactory>();
private readonly static object TRANSPORT_FACTORY_TYPES_LOCK = new
object();
private readonly static IDictionary<String, Type>
TRANSPORT_FACTORY_TYPES = new Dictionary<String, Type>();
Modified: activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/vs2008-mqtt.csproj
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/vs2008-mqtt.csproj?rev=1542552&r1=1542551&r2=1542552&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/vs2008-mqtt.csproj (original)
+++ activemq/activemq-dotnet/Apache.NMS.MQTT/trunk/vs2008-mqtt.csproj Sat Nov
16 19:08:02 2013
@@ -89,6 +89,8 @@
<Compile Include="src\main\csharp\Util\MQTTDestination.cs" />
<Compile Include="src\main\csharp\Transport\TransportFactory.cs" />
<Compile
Include="src\main\csharp\Transport\MQTTTransportFactoryAttribute.cs" />
+ <Compile Include="src\main\csharp\Commands\PUBCOMP.cs" />
+ <Compile Include="src\main\csharp\RequestTimedOutException.cs" />
</ItemGroup>
<ItemGroup>
<Folder Include="keyfile\" />
