Merge pull request #35 from Havret/ping_pong_benchmark
NO-JIRA: Pingpong benchmark
diff --git a/.travis.yml b/.travis.yml
new file mode 100644
index 0000000..7767232
--- /dev/null
+++ b/.travis.yml
@@ -0,0 +1,7 @@
+language: csharp
+solution: apache-nms-amqp.sln
+mono: none
+dotnet: 2.2.401
+script:
+ - dotnet build -p:AppTargetFramework=netcoreapp2.2 -c Release
+ - dotnet test ./test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj -f netcoreapp2.2 -c Release
\ No newline at end of file
diff --git a/README.md b/README.md
index 38e0e7d..6862b99 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,13 @@
# Apache-NMS-AMQP
-### Overview
+
+## Build Status
+
+OS | Status
+---|---
+Linux | [](https://travis-ci.org/apache/activemq-nms-amqp)
+Windows | [](https://ci.appveyor.com/project/ApacheSoftwareFoundation/activemq-nms-amqp)
+
+## Overview
The goal of this project is to combine the [.NET Message Service API](http://activemq.apache.org/nms/) (NMS) with
the [Advanced Message Queuing Protocol (AMQP)](https://www.amqp.org/) 1.0 standard wireline protocol. Historically, the Apache community created the NMS API which provided a vendor agnostic .NET interface to a variety of messaging systems. The NMS API gives the flexibility to write .NET applications in C#, VB or any other .NET language, all while using a single API to connect to any number of messaging providers. The Advanced Message Queuing Protocol (AMQP) is an open and standardized internet protocol for reliably passing messages between applications or organizations.
Before AMQP became a standard, organizations used proprietary wireline protocols to connect their systems which lead to vendor lock-in and integration problems when integrating with external organizations.
@@ -12,57 +20,19 @@
If you are a .NET developer that doesn't want to be locked into a messaging implementation then get engaged with this project. Here you will find the open source code base and please provide comments and make your own enhancements. The project will be folded into the Apache community once fully mature.
-### AMQP1.0 Protocol Engine AmqpNetLite
+## AMQP1.0 Protocol Engine AmqpNetLite
Apache-NMS-AMQP uses [AmqpNetLite](https://github.com/Azure/amqpnetlite) as the underlying AMQP 1.0 transport Protocol engine.
-### Overall Architecture
+## Overall Architecture
Apache-NMS-AMQP should bridge the familiar NMS concepts to AMQP protocol concepts as described in the document [amqp-bindmap-jms-v1.0-wd09.pdf](https://www.oasis-open.org/committees/download.php/60574/amqp-bindmap-jms-v1.0-wd09.pdf).
So in general most of the top level classes that implement the Apache.NMS interface _Connection, Session, MessageProducer,_ etc create, manage, and destroy the amqpnetlite equivalent object _Connection, Session, Link,_ etc.
-### Building With Visual Studio 2019
-There are multiple projects: Apache-NMS-AMQP, Apache-NMS-AMQP.Test, and HelloWorld. All projects use the new csproj format available in Visual Studio 2019.
-Apache-NMS-AMQP is the library which implements The Apache.NMS Interface using AmqpNetLite.
-Apache-NMS-AMQP.Test produces an NUnit dll for unit testing.
-HelloWorld is a sample application using the NMS library which can send messages to an AMQP Message Broker.
+## Getting started
+- [Configuration](docs/configuration.md) contains various uri options that can be set when defining a ConnectionFactory.
+- [Interested in the code?](docs/working_with_code.md) Clone and build the projects.
+- Want to contribute? Github pull requests are one way to contribute, but our real issue tracker is [JIRA](https://issues.apache.org/jira/issues/?jql=project%20%3D%20AMQNET%20AND%20component%20%3D%20AMQP).
-To build, launch Visual Studio 2019 with the nms-amqp.sln file and build the solution.
-Build artifacts will be under `<root_folder>\<project_folder>\bin\$(Configuration)\$(TargetFramework)`.
-
-### Building With DotNet SDK
-Alternatively, to build without Visual Studio 2019 the project can be built using [.NET Core sdk tool](https://www.microsoft.com/net/download/windows), version 2.2.+.
-Execute the dotnet sdk command to build all projects:
-```
-<root_folder>>dotnet build nms-amqp.sln
-```
-
-### Testing
-Tests use the NUnit Framework. The tests include both unit and system tests (require a broker).
-
-Apache-NMS-AMQP-Test contains only unit tests and doesn't require any configuration and dependencies.
-
-Apache-NMS-AMQP-Interop-Test contains system tests and require broker to be up and running. Broker can be configured either directly from the code (to do so you have to edit _AmqpTestSupport_ base class), or using environment variables:
-
-| Variable | Meaning |
-|----------|---------|
-|NMS_AMQP_TEST_URI|Broker uri|
-|NMS_AMQP_TEST_CU|Username|
-|NMS_AMQP_TEST_CPWD|Password|
-
-```
-NUNIT3-CONSOLE Apache-NMS-AMQP-Interop-Test.dll -p:uri=brokerUri -p:cu=userName -p:cpwd=password
-```
-
-#### VS2019 Test Explorer
-Visual Studio 2019 will also run NUnit tests with the built-in TestExplorer tool.
-
-#### dotnet test
-
-If building with the dotnet sdk, From the top level directory simply enter _dotnet test_ to build and run all the tests. Individual tests can be run with:
-```
-dotnet test filter=<Test Name>
-```
-
-### Amqp Provider NMS Feature Support
+## Amqp Provider NMS Feature Support
| Feature | Supported | Comments |
|---------------|:---------:|:-----------------|
@@ -102,4 +72,4 @@
# Licensing
-This software is licensed under the terms you may find in the file named "LICENSE" in this directory.
+This software is licensed under the terms you may find in the file named "LICENSE" in this directory.
\ No newline at end of file
diff --git a/appveyor.yml b/appveyor.yml
new file mode 100644
index 0000000..f8c0bb4
--- /dev/null
+++ b/appveyor.yml
@@ -0,0 +1,10 @@
+image: Visual Studio 2019
+configuration: Release
+test:
+ assemblies:
+ - '**\NMS.AMQP.Test.dll'
+before_build:
+ - cmd: dotnet --version
+ - cmd: dotnet restore apache-nms-amqp.sln
+build_script:
+ - cmd: dotnet build apache-nms-amqp.sln
\ No newline at end of file
diff --git a/docs/configuration.md b/docs/configuration.md
new file mode 100644
index 0000000..2b3046e
--- /dev/null
+++ b/docs/configuration.md
@@ -0,0 +1,69 @@
+# Client configuration
+This file details various configuration options for the client, the syntax for its related configuration, and various uri options that can be set when defining a ConnectionFactory.
+
+## Connection Uri
+The basic format of the clients Connection URI is as follows:
+
+```
+amqp[s]://hostname:port[?option=value[&option2=value...]]
+```
+
+Where the amqps and scheme is specified to use SSL/TLS.
+
+The client can be configured with a number of different settings using the uri while defining the ConnectionFactory, these are detailed in the following sections.
+
+### NMS Configuration options
+The options apply to the behavior of the NMS objects such as Connection, Session, MessageConsumer and MessageProducer.
+
+- **nms.username** User name value used to authenticate the connection.
+- **nms.password** The password value used to authenticate the connection.
+- **nms.clientId** The ClientId value that is applied to the connection.
+- **nms.localMessageExpiry** Controls whether MessageConsumer instances will locally filter expired Messages or deliver them. By default this value is set to true and expired messages will be filtered.
+- **nms.sendTimeout** Timeout value that controls how long the client waits on completion of a synchronous message send before returning an error. By default the client will wait indefinitely for a send to complete.
+- **nms.requestTimeout** Timeout value that controls how long the client waits on completion of various synchronous interactions, such as opening a producer or consumer, before returning an error. Does not affect synchronous message sends. By default the client will wait indefinitely for a request to complete.
+- **nms.clientIdPrefix** Optional prefix value that is used for generated Client ID values when a new Connection is created for the JMS ConnectionFactory. The default prefix is 'ID:'.
+- **nms.connectionIdPrefix** Optional prefix value that is used for generated Connection ID values when a new Connection is created for the JMS ConnectionFactory. This connection ID is used when logging some information from the JMS Connection object so a configurable prefix can make breadcrumbing the logs easier. The default prefix is 'ID:'.
+
+### TCP Transport Configuration options
+When connected to a remote using plain TCP these options configure the behaviour of the underlying socket. These options are appended to the connection URI along with the other configuration options, for example:
+
+```
+amqp://localhost:5672?nms.clientId=foo&transport.receiveBufferSize=30000
+```
+
+The complete set of TCP Transport options is listed below:
+
+- **transport.sendBufferSize** Specifies the ReceiveBufferSize option of the TCP socket.
+- **transport.receiveBufferSize** Specifies the SendBufferSize option of the TCP socket.
+- **transport.receiveTimeout** Specifies the ReceiveTimeout option of the TCP socket.
+- **transport.sendTimeout** Specifies the SendTimeout option of the TCP socket.
+- **transport.tcpKeepAliveTime** Specifies how often a keep-alive transmission is sent to an idle connection.
+- **transport.tcpKeepAliveInterval** Specifies how often a keep-alive transmission is sent when no response is received from previous keep-alive transmissions.
+- **transport.tcpNoDelay** Specifies the NoDelay option of the TCP socket.
+
+If *tcpKeepAliveTime* or *tcpKeepAliveInterval* it set, TCP Keep-Alive is enabled.
+
+### Failover Configuration options
+With failover enabled the client can reconnect to another server automatically when connection to the current server is lost for some reason. The failover URI is always initiated with the failover prefix and a list of URIs for the server(s) is contained inside a set of parentheses. The "nms." options are applied to the overall failover URI, outside the parentheses, and affect the NMS Connection object for its lifetime.
+
+The URI for failover looks something like the following:
+
+```
+failover:(amqp://host1:5672,amqp://host2:5672)?nms.clientId=foo&failover.maxReconnectAttempts=20
+```
+
+The individual broker details within the parentheses can use the "transport." or "amqp." options defined earlier, with these being applied as each host is connected to:
+
+```
+failover:(amqp://host1:5672?amqp.option=value,amqp://host2:5672?transport.option=value)?nms.clientId=foo
+```
+
+The complete set of configuration options for failover is listed below:
+
+- **failover.initialReconnectDelay** The amount of time the client will wait before the first attempt to reconnect to a remote peer. The default value is zero, meaning the first attempt happens immediately.
+- **failover.reconnectDelay** Controls the delay between successive reconnection attempts, defaults to 10 milliseconds. If the backoff option is not enabled this value remains constant.
+- **failover.maxReconnectDelay** The maximum time that the client will wait before attempting a reconnect. This value is only used when the backoff feature is enabled to ensure that the delay doesn't not grow too large. Defaults to 30 seconds as the max time between connect attempts.
+- **failover.useReconnectBackOff** Controls whether the time between reconnection attempts should grow based on a configured multiplier. This option defaults to true.
+- **failover.reconnectBackOffMultiplier** The multiplier used to grow the reconnection delay value, defaults to 2.0d.
+- **failover.maxReconnectAttempts** The number of reconnection attempts allowed before reporting the connection as failed to the client. The default is no limit or (-1).
+- **failover.startupMaxReconnectAttempts** For a client that has never connected to a remote peer before this option control how many attempts are made to connect before reporting the connection as failed. The default is to use the value of maxReconnectAttempts.
\ No newline at end of file
diff --git a/docs/working_with_code.md b/docs/working_with_code.md
new file mode 100644
index 0000000..70d7ec2
--- /dev/null
+++ b/docs/working_with_code.md
@@ -0,0 +1,36 @@
+# Prerequisites
+- Visual Studio 2019. [Community Edition](https://visualstudio.microsoft.com/downloads/) works.
+
+# Build the projects
+- Build with Visual Studio. Open apache-nms-amqp in Visual Studio. This solution contains both source code and test projects for all supported platforms. If the SDK of a particular platform is not present, the project will fail to load. You can either install the required SDK (if available) or remove the project(s) from solution.
+
+- Alternatively, to build without Visual Studio the solution can be built using [.NET Core SDK tool](https://www.microsoft.com/net/download/windows), version 2.2.+.
+
+Execute the dotnet sdk command to build all projects:
+```
+<root_folder>>dotnet build nms-amqp.sln
+```
+
+# Run the tests
+Tests use the NUnit Framework. The tests include both unit and system tests (require a broker).
+
+Apache-NMS-AMQP-Test contains only unit tests and doesn't require any configuration and dependencies.
+
+Apache-NMS-AMQP-Interop-Test contains system tests and require broker to be up and running. Broker can be configured either directly from the code (to do so you have to edit _AmqpTestSupport_ base class), or using environment variables:
+
+| Variable | Meaning |
+|----------|---------|
+|NMS_AMQP_TEST_URI|Broker uri|
+|NMS_AMQP_TEST_CU|Username|
+|NMS_AMQP_TEST_CPWD|Password|
+
+## Visual Studio Test Explorer
+Visual Studio 2019 will run NUnit tests with the built-in TestExplorer tool.
+
+## dotnet test
+
+If building with the dotnet sdk, from the top level directory simply enter _dotnet test_ to build and run all the tests. Individual tests can be run with:
+
+```
+dotnet test filter=<Test Name>
+```
\ No newline at end of file
diff --git a/src/HelloWorld/HelloWorld.csproj b/src/HelloWorld/HelloWorld.csproj
index e2176fe..3e156c1 100644
--- a/src/HelloWorld/HelloWorld.csproj
+++ b/src/HelloWorld/HelloWorld.csproj
@@ -18,6 +18,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net462;netcoreapp2.2</TargetFrameworks>
+ <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
<OutputType>Exe</OutputType>
<RootNamespace>HelloWorld</RootNamespace>
<AssemblyName>HelloWorld</AssemblyName>
diff --git a/src/NMS.AMQP/Message/Facade/INmsMessageFacade.cs b/src/NMS.AMQP/Message/Facade/INmsMessageFacade.cs
index 4b855ff..f6e7a94 100644
--- a/src/NMS.AMQP/Message/Facade/INmsMessageFacade.cs
+++ b/src/NMS.AMQP/Message/Facade/INmsMessageFacade.cs
@@ -38,7 +38,7 @@
string GroupId { get; set; }
uint GroupSequence { get; set; }
DateTime? Expiration { get; set; }
- sbyte JmsMsgType { get; }
+ sbyte? JmsMsgType { get; }
/// <summary>
/// True if this message is tagged as being persistent
diff --git a/src/NMS.AMQP/NmsConnection.cs b/src/NMS.AMQP/NmsConnection.cs
index f8bf283..a34d343 100644
--- a/src/NMS.AMQP/NmsConnection.cs
+++ b/src/NMS.AMQP/NmsConnection.cs
@@ -314,10 +314,14 @@
{
session.OnInboundMessage(envelope);
}
+ else
+ {
+ Tracer.Error($"Could not dispatch message {envelope.Message.NMSMessageId} because session {envelope.ConsumerInfo.SessionId} not found.");
+ }
if (connectionListeners.Any())
{
- foreach (INmsConnectionListener listener in connectionListeners)
+ foreach (INmsConnectionListener listener in connectionListeners)
listener.OnInboundMessage(envelope.Message);
}
}
diff --git a/src/NMS.AMQP/NmsMessageConsumer.cs b/src/NMS.AMQP/NmsMessageConsumer.cs
index 34afd0c..b4cf301 100644
--- a/src/NMS.AMQP/NmsMessageConsumer.cs
+++ b/src/NMS.AMQP/NmsMessageConsumer.cs
@@ -27,7 +27,7 @@
{
public class NmsMessageConsumer : IMessageConsumer
{
- private static readonly object SyncRoot = new object();
+ private readonly object syncRoot = new object();
private readonly AcknowledgementMode acknowledgementMode;
private readonly AtomicBool closed = new AtomicBool();
private readonly MessageDeliveryTask deliveryTask;
@@ -92,7 +92,7 @@
if (closed)
return;
- lock (SyncRoot)
+ lock (syncRoot)
{
Shutdown(null);
Session.Connection.DestroyResource(Info).ConfigureAwait(false).GetAwaiter().GetResult();
@@ -106,7 +106,7 @@
add
{
CheckClosed();
- lock (SyncRoot)
+ lock (syncRoot)
{
Listener += value;
DrainMessageQueueToListener();
@@ -114,7 +114,7 @@
}
remove
{
- lock (SyncRoot)
+ lock (syncRoot)
{
Listener -= value;
}
@@ -200,6 +200,11 @@
public void OnInboundMessage(InboundMessageDispatch envelope)
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug($"Message {envelope.Message.NMSMessageId} passed to consumer {Info.Id}");
+ }
+
SetAcknowledgeCallback(envelope);
if (envelope.EnqueueFirst)
@@ -215,9 +220,14 @@
private void DeliverNextPending()
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug($"{Info.Id} is about to deliver next pending message.");
+ }
+
if (Session.IsStarted && started && Listener != null)
{
- lock (SyncRoot)
+ lock (syncRoot)
{
try
{
@@ -225,19 +235,30 @@
{
var envelope = messageQueue.DequeueNoWait();
if (envelope == null)
+ {
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug($"No message available for delivery.");
+ }
+
return;
+ }
if (IsMessageExpired(envelope))
{
if (Tracer.IsDebugEnabled)
+ {
Tracer.Debug($"{Info.Id} filtered expired message: {envelope.Message.NMSMessageId}");
+ }
DoAckExpired(envelope);
}
else if (IsRedeliveryExceeded(envelope))
{
if (Tracer.IsDebugEnabled)
- Tracer.Debug($"{Info.Id} filtered message with excessive redelivery count: {envelope.RedeliveryCount}");
+ {
+ Tracer.Debug($"{Info.Id} filtered message with excessive redelivery count: {envelope.RedeliveryCount.ToString()}");
+ }
// TODO: Apply redelivery policy
DoAckExpired(envelope);
@@ -320,6 +341,11 @@
while (true)
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug("Trying to dequeue next message.");
+ }
+
InboundMessageDispatch envelope = messageQueue.Dequeue(timeout);
if (failureCause != null)
@@ -331,7 +357,9 @@
if (IsMessageExpired(envelope))
{
if (Tracer.IsDebugEnabled)
+ {
Tracer.Debug($"{Info.Id} filtered expired message: {envelope.Message.NMSMessageId}");
+ }
DoAckExpired(envelope);
@@ -341,7 +369,9 @@
else if (IsRedeliveryExceeded(envelope))
{
if (Tracer.IsDebugEnabled)
- Tracer.Debug($"{Info.Id} filtered message with excessive redelivery count: {envelope.RedeliveryCount}");
+ {
+ Tracer.Debug($"{Info.Id} filtered message with excessive redelivery count: {envelope.RedeliveryCount.ToString()}");
+ }
// TODO: Apply redelivery policy
DoAckExpired(envelope);
@@ -349,7 +379,9 @@
else
{
if (Tracer.IsDebugEnabled)
- Tracer.Debug($"{Info.Id} received message {envelope.Message.NMSMessageId}.");
+ {
+ Tracer.Debug($"{Info.Id} received message {envelope.Message.NMSMessageId}.");
+ }
AckFromReceive(envelope);
return envelope.Message.Copy();
@@ -408,7 +440,7 @@
{
envelope.Message.NmsAcknowledgeCallback = new NmsAcknowledgeCallback(Session);
}
- else if (Session.AcknowledgementMode == AcknowledgementMode.IndividualAcknowledge)
+ else if (Session.IsIndividualAcknowledge())
{
envelope.Message.NmsAcknowledgeCallback = new NmsAcknowledgeCallback(Session, envelope);
}
@@ -423,11 +455,23 @@
{
if (closed.CompareAndSet(false, true))
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug("Shutting down NmsMessageConsumer.");
+ }
+
failureCause = exception;
Session.Remove(this);
started.Set(false);
messageQueue.Dispose();
}
+ else
+ {
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug("NmsMessageConsumer already closed.");
+ }
+ }
}
public void Start()
@@ -463,7 +507,7 @@
public void Stop()
{
- lock (SyncRoot)
+ lock (syncRoot)
{
started.Set(false);
}
diff --git a/src/NMS.AMQP/NmsSession.cs b/src/NMS.AMQP/NmsSession.cs
index c9f008d..9904098 100644
--- a/src/NMS.AMQP/NmsSession.cs
+++ b/src/NMS.AMQP/NmsSession.cs
@@ -344,6 +344,10 @@
{
messageConsumer.OnInboundMessage(envelope);
}
+ else
+ {
+ Tracer.Error($"Could not dispatch message {envelope.Message.NMSMessageId} because consumer {envelope.ConsumerId} not found.");
+ }
}
public void Acknowledge(AckType ackType)
@@ -422,6 +426,11 @@
internal void EnqueueForDispatch(NmsMessageConsumer.MessageDeliveryTask task)
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug("Message enqueued for dispatch.");
+ }
+
dispatcher?.Post(task);
}
@@ -451,7 +460,7 @@
public NmsMessageProducer ProducerClosed(Id producerId, Exception cause)
{
- Tracer.Info($"A NMS MessageProducer has been closed: {producerId}");
+ Tracer.Info($"A NmsMessageProducer has been closed: {producerId}. Cause: {cause}");
if (producers.TryGetValue(producerId, out NmsMessageProducer producer))
{
@@ -464,6 +473,14 @@
Tracer.DebugFormat("Ignoring exception thrown during cleanup of closed producer", error);
}
}
+ else
+ {
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug($"NmsMessageProducer: {producerId} not found in session {this.SessionInfo.Id}.");
+ }
+
+ }
return producer;
}
@@ -592,5 +609,7 @@
consumer.OnConnectionInterrupted();
}
}
+
+ internal bool IsIndividualAcknowledge() => acknowledgementMode == AcknowledgementMode.IndividualAcknowledge;
}
}
\ No newline at end of file
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs
index fb4fd37..1481248 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs
@@ -195,6 +195,11 @@
private void OnMessage(IReceiverLink receiver, global::Amqp.Message amqpMessage)
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug($"Received message {amqpMessage?.Properties?.MessageId}");
+ }
+
NmsMessage message;
try
{
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpProvider.cs b/src/NMS.AMQP/Provider/Amqp/AmqpProvider.cs
index 24022a8..99330db 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpProvider.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpProvider.cs
@@ -41,6 +41,66 @@
this.transport = transport;
}
+ static AmqpProvider()
+ {
+ // Set up tracing in AMQP. We capture all AMQP traces in the TraceListener below
+ // and map to NMS 'Tracer' logs as follows:
+ // AMQP Tracer
+ // Verbose Debug
+ // Frame Debug
+ // Information Info
+ // Output Info (should not happen)
+ // Warning Warn
+ // Error Error
+ Trace.TraceLevel = TraceLevel.Verbose;
+ Trace.TraceListener = (level, format, args) =>
+ {
+ switch (level)
+ {
+ case TraceLevel.Verbose:
+ case TraceLevel.Frame:
+ Tracer.DebugFormat(format, args);
+ break;
+ case TraceLevel.Information:
+ case TraceLevel.Output:
+ //
+ // Applications should not access AmqpLite directly so there
+ // should be no 'Output' level logs.
+ Tracer.InfoFormat(format, args);
+ break;
+ case TraceLevel.Warning:
+ Tracer.WarnFormat(format, args);
+ break;
+ case TraceLevel.Error:
+ Tracer.ErrorFormat(format, args);
+ break;
+ default:
+ Tracer.InfoFormat("Unknown AMQP LogLevel: {}", level);
+ Tracer.InfoFormat(format, args);
+ break;
+ }
+ };
+ }
+
+ /// <summary>
+ /// Enables AmqpNetLite's Frame logging level.
+ /// </summary>
+ public bool TraceFrames
+ {
+ get => ((Trace.TraceLevel & TraceLevel.Frame) == TraceLevel.Frame);
+ set
+ {
+ if (value)
+ {
+ Trace.TraceLevel = Trace.TraceLevel | TraceLevel.Frame;
+ }
+ else
+ {
+ Trace.TraceLevel = Trace.TraceLevel & ~TraceLevel.Frame;
+ }
+ }
+ }
+
public long SendTimeout => connectionInfo?.SendTimeout ?? ConnectionInfo.DEFAULT_SEND_TIMEOUT;
public Uri RemoteUri { get; }
public IProviderListener Listener { get; private set; }
@@ -58,6 +118,11 @@
internal void OnConnectionClosed(Error error)
{
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug($"Connection closed. {error}");
+ }
+
bool connectionExplicitlyClosed = error == null;
if (!connectionExplicitlyClosed)
{
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpCodec.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpCodec.cs
index f371610..299657f 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpCodec.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpCodec.cs
@@ -28,7 +28,7 @@
public static INmsMessageFacade DecodeMessage(IAmqpConsumer consumer, global::Amqp.Message amqpMessage)
{
// First we try the easy way, if the annotation is there we don't have to work hard.
- AmqpNmsMessageFacade result = CreateFromMsgAnnotation(amqpMessage.MessageAnnotations);
+ AmqpNmsMessageFacade result = CreateFromMsgAnnotation(amqpMessage);
if (result == null)
{
// Next, match specific section structures and content types
@@ -44,9 +44,9 @@
throw new NMSException("Could not create a NMS message from incoming message");
}
- private static AmqpNmsMessageFacade CreateFromMsgAnnotation(MessageAnnotations messageAnnotations)
+ private static AmqpNmsMessageFacade CreateFromMsgAnnotation(global::Amqp.Message message)
{
- object annotation = messageAnnotations?[SymbolUtil.JMSX_OPT_MSG_TYPE];
+ object annotation = message.MessageAnnotations?[SymbolUtil.JMSX_OPT_MSG_TYPE];
if (annotation != null)
{
@@ -55,16 +55,19 @@
{
case MessageSupport.JMS_TYPE_MSG:
return new AmqpNmsMessageFacade();
- case MessageSupport.JMS_TYPE_BYTE:
- return new AmqpNmsBytesMessageFacade();
case MessageSupport.JMS_TYPE_TXT:
return new AmqpNmsTextMessageFacade();
- case MessageSupport.JMS_TYPE_OBJ:
- return new AmqpNmsObjectMessageFacade();
case MessageSupport.JMS_TYPE_STRM:
return new AmqpNmsStreamMessageFacade();
case MessageSupport.JMS_TYPE_MAP:
return new AmqpNmsMapMessageFacade();
+ case MessageSupport.JMS_TYPE_BYTE:
+ // Java serialized objects should be treated as bytes messages
+ // as we cannot deserialize them in .NET world
+ case MessageSupport.JMS_TYPE_OBJ when IsContentType(SymbolUtil.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE, GetContentType(message.Properties)):
+ return new AmqpNmsBytesMessageFacade();
+ case MessageSupport.JMS_TYPE_OBJ:
+ return new AmqpNmsObjectMessageFacade();
default:
throw new NMSException("Invalid Message Type annotation value found in message: " + annotation);
}
@@ -130,9 +133,10 @@
{
if (messageFacade.Message.MessageAnnotations == null)
messageFacade.Message.MessageAnnotations = new MessageAnnotations();
-
- messageFacade.Message.MessageAnnotations[SymbolUtil.JMSX_OPT_MSG_TYPE] = messageFacade.JmsMsgType;
-
+
+ if (messageFacade.JmsMsgType.HasValue)
+ messageFacade.Message.MessageAnnotations[SymbolUtil.JMSX_OPT_MSG_TYPE] = messageFacade.JmsMsgType.Value;
+
AmqpDestinationHelper.SetToAnnotationFromDestination(messageFacade.NMSDestination, messageFacade.Message.MessageAnnotations);
AmqpDestinationHelper.SetReplyToAnnotationFromDestination(messageFacade.NMSReplyTo, messageFacade.Message.MessageAnnotations);
}
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsBytesMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsBytesMessageFacade.cs
index 84e167e..e61ddfb 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsBytesMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsBytesMessageFacade.cs
@@ -33,7 +33,7 @@
private static readonly Data EMPTY_DATA = new Data { Binary = new byte[0] };
- public override sbyte JmsMsgType => MessageSupport.JMS_TYPE_BYTE;
+ public override sbyte? JmsMsgType => MessageSupport.JMS_TYPE_BYTE;
public long BodyLength => GetBinaryFromBody().Binary.LongLength;
public BinaryReader GetDataReader()
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMapMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMapMessageFacade.cs
index 8be31cf..ae09084 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMapMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMapMessageFacade.cs
@@ -29,7 +29,7 @@
private AMQPValueMap map;
public IPrimitiveMap Map => map;
- public override sbyte JmsMsgType => MessageSupport.JMS_TYPE_MAP;
+ public override sbyte? JmsMsgType => MessageSupport.JMS_TYPE_MAP;
protected override void InitializeEmptyBody()
{
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs
index 6766602..183f215 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs
@@ -333,7 +333,7 @@
}
public MessageAnnotations MessageAnnotations => Message.MessageAnnotations;
- public virtual sbyte JmsMsgType => MessageSupport.JMS_TYPE_MSG;
+ public virtual sbyte? JmsMsgType => MessageSupport.JMS_TYPE_MSG;
/// <summary>
/// The annotation value for the JMS Message content type. For a generic JMS message this
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsObjectMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsObjectMessageFacade.cs
index 115521c..3e64d71 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsObjectMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsObjectMessageFacade.cs
@@ -35,7 +35,7 @@
set => Delegate.Object = value;
}
- public override sbyte JmsMsgType => MessageSupport.JMS_TYPE_OBJ;
+ public override sbyte? JmsMsgType => typeDelegate is AmqpTypedObjectDelegate ? (sbyte?) MessageSupport.JMS_TYPE_OBJ : null;
public override void OnSend(TimeSpan producerTtl)
{
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsStreamMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsStreamMessageFacade.cs
index 509f7ef..18fc051 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsStreamMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsStreamMessageFacade.cs
@@ -30,7 +30,7 @@
private IList list;
private int position = 0;
- public override sbyte JmsMsgType => MessageSupport.JMS_TYPE_STRM;
+ public override sbyte? JmsMsgType => MessageSupport.JMS_TYPE_STRM;
public object Peek()
{
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsTextMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsTextMessageFacade.cs
index 01610d6..dd3cf81 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsTextMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsTextMessageFacade.cs
@@ -83,7 +83,7 @@
}
}
- public override sbyte JmsMsgType => MessageSupport.JMS_TYPE_TXT;
+ public override sbyte? JmsMsgType => MessageSupport.JMS_TYPE_TXT;
private static string DecodeBinaryBody(byte[] body)
{
diff --git a/src/NMS.AMQP/Transport/ISecureTransportContext.cs b/src/NMS.AMQP/Transport/ISecureTransportContext.cs
index 4cac12e..97762dd 100644
--- a/src/NMS.AMQP/Transport/ISecureTransportContext.cs
+++ b/src/NMS.AMQP/Transport/ISecureTransportContext.cs
@@ -21,9 +21,6 @@
{
public interface ISecureTransportContext : ITransportContext
{
-
- new ISecureTransportContext Copy();
-
string ServerName { get; set; }
string ClientCertFileName { get; set; }
diff --git a/src/NMS.AMQP/Transport/ITransportContext.cs b/src/NMS.AMQP/Transport/ITransportContext.cs
index ad225be..05eec6e 100644
--- a/src/NMS.AMQP/Transport/ITransportContext.cs
+++ b/src/NMS.AMQP/Transport/ITransportContext.cs
@@ -31,13 +31,21 @@
int SendBufferSize { get; set; }
int SendTimeout { get; set; }
+
bool TcpNoDelay { get; set; }
- bool UseLogging { get; set; }
+ /// <summary>
+ /// Gets or sets how often a keep-alive transmission is sent to an idle connection.
+ /// </summary>
+ uint TcpKeepAliveTime { get; set; }
+
+ /// <summary>
+ /// Gets or sets How often a keep-alive transmission is sent when no response is received from previous keep-alive transmissions.
+ /// </summary>
+ uint TcpKeepAliveInterval { get; set; }
+
bool IsSecure { get; }
-
- ITransportContext Copy();
-
+
Task<Connection> CreateAsync(Address address, IHandler handler);
}
}
diff --git a/src/NMS.AMQP/Transport/SecureTransportContext.cs b/src/NMS.AMQP/Transport/SecureTransportContext.cs
index cc3913c..08cfc66 100644
--- a/src/NMS.AMQP/Transport/SecureTransportContext.cs
+++ b/src/NMS.AMQP/Transport/SecureTransportContext.cs
@@ -37,8 +37,8 @@
internal class SecureTransportContext : TransportContext, ISecureTransportContext
{
- private readonly static List<string> SupportedProtocols;
- private readonly static Dictionary<string, int> SupportedProtocolValues;
+ private static readonly List<string> SupportedProtocols;
+ private static readonly Dictionary<string, int> SupportedProtocolValues;
#region static Initializer
@@ -411,65 +411,6 @@
}
#endregion
-
- #region Copy Methods
-
-
- protected override void CopyBuilder(Amqp.ConnectionFactory copy)
- {
- base.CopyBuilder(copy);
-
- copy.SSL.Protocols = connectionBuilder.SSL.Protocols;
- copy.SSL.CheckCertificateRevocation = connectionBuilder.SSL.CheckCertificateRevocation;
-
- if (connectionBuilder.SSL.ClientCertificates != null)
- {
- copy.SSL.ClientCertificates = new X509CertificateCollection(connectionBuilder.SSL.ClientCertificates);
- }
-
- }
-
- protected override void CopyInto(TransportContext copy)
- {
- SecureTransportContext stcCopy = copy as SecureTransportContext;
-
- // Copy Secure properties.
-
- // copy keystore properties
- stcCopy.KeyStoreName = this.KeyStoreName;
- stcCopy.KeyStorePassword = this.KeyStorePassword;
- stcCopy.KeyStoreLocation = this.KeyStoreLocation;
-
- // copy certificate properties
- stcCopy.AcceptInvalidBrokerCert = this.AcceptInvalidBrokerCert;
- stcCopy.ServerName = this.ServerName;
- stcCopy.ClientCertFileName = this.ClientCertFileName;
- stcCopy.ClientCertPassword = this.ClientCertPassword;
- stcCopy.ClientCertSubject = this.ClientCertSubject;
-
- // copy application callback
- stcCopy.ServerCertificateValidateCallback = this.ServerCertificateValidateCallback;
- stcCopy.ClientCertificateSelectCallback = this.ClientCertificateSelectCallback;
-
- base.CopyInto(copy);
-
- stcCopy.connectionBuilder.SSL.RemoteCertificateValidationCallback = this.ContextServerCertificateValidation;
- stcCopy.connectionBuilder.SSL.LocalCertificateSelectionCallback = this.ContextLocalCertificateSelect;
- }
-
- public override ITransportContext Copy()
- {
- TransportContext copy = new SecureTransportContext();
- this.CopyInto(copy);
- return copy;
- }
-
- ISecureTransportContext ISecureTransportContext.Copy()
- {
- return this.Copy() as SecureTransportContext;
- }
-
- #endregion
}
}
diff --git a/src/NMS.AMQP/Transport/TransportContext.cs b/src/NMS.AMQP/Transport/TransportContext.cs
index 229d8e7..e62c9ac 100644
--- a/src/NMS.AMQP/Transport/TransportContext.cs
+++ b/src/NMS.AMQP/Transport/TransportContext.cs
@@ -26,9 +26,6 @@
namespace Apache.NMS.AMQP.Transport
{
-
- #region Transport Context
-
/// <summary>
/// Transport management is mainly handled by the AmqpNetLite library, Except for custom transports.
/// TransportContext should configure the Amqp.ConnectionFactory for the tcp transport properties.
@@ -43,58 +40,25 @@
connectionBuilder.SASL.Profile = Amqp.Sasl.SaslProfile.Anonymous;
}
- static TransportContext()
- {
- //
- // Set up tracing in AMQP. We capture all AMQP traces in the TraceListener below
- // and map to NMS 'Tracer' logs as follows:
- // AMQP Tracer
- // Verbose Debug
- // Frame Debug
- // Information Info
- // Output Info (should not happen)
- // Warning Warn
- // Error Error
- //
- Amqp.Trace.TraceLevel = Amqp.TraceLevel.Verbose | Amqp.TraceLevel.Frame;
- Amqp.Trace.TraceListener = (level, format, args) =>
- {
- switch (level)
- {
- case Amqp.TraceLevel.Verbose:
- case Amqp.TraceLevel.Frame:
- Tracer.DebugFormat(format, args);
- break;
- case Amqp.TraceLevel.Information:
- case Amqp.TraceLevel.Output:
- //
- // Applications should not access AmqpLite directly so there
- // should be no 'Output' level logs.
- Tracer.InfoFormat(format, args);
- break;
- case Amqp.TraceLevel.Warning:
- Tracer.WarnFormat(format, args);
- break;
- case Amqp.TraceLevel.Error:
- Tracer.ErrorFormat(format, args);
- break;
- default:
- Tracer.InfoFormat("Unknown AMQP LogLevel: {}", level);
- Tracer.InfoFormat(format, args);
- break;
- }
- };
- }
-
- #region Transport Options
-
public int ReceiveBufferSize { get => this.connectionBuilder.TCP.ReceiveBufferSize; set => this.connectionBuilder.TCP.ReceiveBufferSize = value; }
public int ReceiveTimeout { get => this.connectionBuilder.TCP.ReceiveTimeout; set => this.connectionBuilder.TCP.ReceiveTimeout = value; }
public int SendBufferSize { get => this.connectionBuilder.TCP.SendBufferSize; set => this.connectionBuilder.TCP.SendBufferSize = value; }
public int SendTimeout { get => this.connectionBuilder.TCP.SendTimeout; set => this.connectionBuilder.TCP.SendTimeout = value; }
public bool TcpNoDelay { get => this.connectionBuilder.TCP.NoDelay; set => this.connectionBuilder.TCP.NoDelay = value; }
- public uint TcpKeepAliveTime { get => this.connectionBuilder.TCP.KeepAlive.KeepAliveTime; set => this.connectionBuilder.TCP.KeepAlive.KeepAliveTime = value; }
- public uint TcpKeepAliveInterval { get => this.connectionBuilder.TCP.KeepAlive.KeepAliveInterval; set => this.connectionBuilder.TCP.KeepAlive.KeepAliveInterval = value; }
+
+ public uint TcpKeepAliveTime
+ {
+ get => this.connectionBuilder.TCP.KeepAlive?.KeepAliveTime ?? default;
+ set => this.TcpKeepAliveSettings.KeepAliveTime = value;
+ }
+
+ public uint TcpKeepAliveInterval
+ {
+ get => this.connectionBuilder.TCP.KeepAlive?.KeepAliveInterval ?? default;
+ set => this.TcpKeepAliveSettings.KeepAliveInterval = value;
+ }
+
+ private TcpKeepAliveSettings TcpKeepAliveSettings => this.connectionBuilder.TCP.KeepAlive ?? (this.connectionBuilder.TCP.KeepAlive = new TcpKeepAliveSettings());
public bool SocketLingerEnabled
{
@@ -128,60 +92,11 @@
}
}
- /// <summary>
- /// UseLogging Enables AmqpNetLite's Frame logging level.
- /// </summary>
- public bool UseLogging
- {
- get => ((Amqp.Trace.TraceLevel & Amqp.TraceLevel.Frame) == Amqp.TraceLevel.Frame);
- set
- {
- if (value)
- {
- Amqp.Trace.TraceLevel = Amqp.Trace.TraceLevel | Amqp.TraceLevel.Frame;
- }
- else
- {
- Amqp.Trace.TraceLevel = Amqp.Trace.TraceLevel & ~Amqp.TraceLevel.Frame;
- }
- }
- }
-
- #endregion
-
public virtual bool IsSecure { get; } = false;
- public virtual ITransportContext Copy()
- {
- TransportContext copy = new TransportContext();
- this.CopyInto(copy);
- return copy;
- }
-
public virtual Task<Connection> CreateAsync(Address address, IHandler handler)
{
return connectionBuilder.CreateAsync(address, handler);
}
-
- protected virtual void CopyInto(TransportContext copy)
- {
- //copy.factory = this.factory;
- //copy.UseLogging = this.UseLogging;
- //Amqp.ConnectionFactory builder = new Amqp.ConnectionFactory();
- //this.CopyBuilder(builder);
- //copy.connectionBuilder = builder;
- }
-
- protected virtual void CopyBuilder(Amqp.ConnectionFactory copy)
- {
- StringDictionary amqpProperties = PropertyUtil.GetProperties(this.connectionBuilder.AMQP);
- StringDictionary tcpProperties = PropertyUtil.GetProperties(this.connectionBuilder.TCP);
- PropertyUtil.SetProperties(copy.AMQP, amqpProperties);
- PropertyUtil.SetProperties(copy.TCP, tcpProperties);
- copy.SASL.Profile = this.connectionBuilder.SASL.Profile;
- }
}
-
- #endregion
-
}
diff --git a/src/NMS.AMQP/Util/IdGenerator.cs b/src/NMS.AMQP/Util/IdGenerator.cs
index 1c91fbf..f3d1c7a 100644
--- a/src/NMS.AMQP/Util/IdGenerator.cs
+++ b/src/NMS.AMQP/Util/IdGenerator.cs
@@ -125,7 +125,6 @@
public void Add(object component)
{
- Tracer.DebugFormat("Adding Component To Id, component {0}, current index: {1}", component, current);
if (isReadOnly)
{
throw new NMSException("Invalid Operation when generating Component Id. Can not change id once generated.");
diff --git a/src/StructuredMessage/StructuredMessage.csproj b/src/StructuredMessage/StructuredMessage.csproj
index 8a210e6..c7c0500 100644
--- a/src/StructuredMessage/StructuredMessage.csproj
+++ b/src/StructuredMessage/StructuredMessage.csproj
@@ -18,6 +18,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net462;netcoreapp2.2</TargetFrameworks>
+ <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
<OutputType>Exe</OutputType>
<RootNamespace>StructuredMessage</RootNamespace>
<AssemblyName>StructuredMessage</AssemblyName>
diff --git a/src/Transactions/Transactions.csproj b/src/Transactions/Transactions.csproj
index d3dd0b3..51f9be8 100644
--- a/src/Transactions/Transactions.csproj
+++ b/src/Transactions/Transactions.csproj
@@ -19,6 +19,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>netcoreapp2.2</TargetFramework>
+ <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
</PropertyGroup>
<ItemGroup>
diff --git a/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj b/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj
index 9e752b9..f7ce890 100644
--- a/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj
+++ b/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj
@@ -17,8 +17,9 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net462;netcoreapp2.2</TargetFrameworks>
+ <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
<RootNamespace>NMS.AMQP.Test</RootNamespace>
- <AssemblyName>NMS.AMQP.Test</AssemblyName>
+ <AssemblyName>NMS.AMQP.Interop.Test</AssemblyName>
<LangVersion>7.3</LangVersion>
</PropertyGroup>
diff --git a/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj b/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj
index 7e5cfa1..68c6937 100644
--- a/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj
+++ b/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj
@@ -17,6 +17,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net462;netcoreapp2.2</TargetFrameworks>
+ <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
<RootNamespace>NMS.AMQP.Test</RootNamespace>
<AssemblyName>NMS.AMQP.Test</AssemblyName>
<LangVersion>7.3</LangVersion>
@@ -30,6 +31,7 @@
<ItemGroup>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.1.1" />
<PackageReference Include="Moq" Version="4.11.0" />
+ <PackageReference Include="NLog" Version="4.6.7" />
<PackageReference Include="NUnit" Version="3.12.0" />
<PackageReference Include="NUnit.Console" Version="3.10.0" />
<PackageReference Include="NUnit3TestAdapter" Version="3.13.0" />
@@ -43,6 +45,9 @@
<None Include=".\config\cert\client.crt" Link="client.crt">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</None>
+ <None Update="NLog.config">
+ <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+ </None>
<None Update="TestSuite.config">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</None>
diff --git a/test/Apache-NMS-AMQP-Test/ConnectionFactoryTest.cs b/test/Apache-NMS-AMQP-Test/ConnectionFactoryTest.cs
index b32fa4e..229b62a 100644
--- a/test/Apache-NMS-AMQP-Test/ConnectionFactoryTest.cs
+++ b/test/Apache-NMS-AMQP-Test/ConnectionFactoryTest.cs
@@ -76,7 +76,8 @@
"&nms.connectionIdPrefix=ID:TEST" +
"&nms.clientIDPrefix=clientId" +
"&nms.requestTimeout=1000" +
- "&nms.sendTimeout=1000";
+ "&nms.sendTimeout=1000" +
+ "&nms.localMessageExpiry=false";
NmsConnectionFactory factory = new NmsConnectionFactory(configured);
@@ -87,6 +88,7 @@
Assert.AreEqual("clientId", factory.ClientIdPrefix);
Assert.AreEqual(1000, factory.RequestTimeout);
Assert.AreEqual(1000, factory.SendTimeout);
+ Assert.IsFalse(factory.LocalMessageExpiry);
}
[Test]
diff --git a/test/Apache-NMS-AMQP-Test/Integration/AmqpAcknowledgmentsIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/AmqpAcknowledgmentsIntegrationTest.cs
index 09bdc04..ba09d9c 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/AmqpAcknowledgmentsIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/AmqpAcknowledgmentsIntegrationTest.cs
@@ -296,7 +296,7 @@
testPeer.ExpectDispositionThatIsAcceptedAndSettled();
for (int i = 0; i < msgCount; i++)
- Assert.NotNull(consumer.Receive(TimeSpan.FromMilliseconds(3000)));
+ Assert.NotNull(consumer.Receive(TimeSpan.FromMilliseconds(3000)), $"Message {i} not received within given timeout.");
testPeer.WaitForAllMatchersToComplete(3000);
diff --git a/test/Apache-NMS-AMQP-Test/Integration/ConnectionIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/ConnectionIntegrationTest.cs
index e9849a5..ec86699 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/ConnectionIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/ConnectionIntegrationTest.cs
@@ -123,7 +123,7 @@
IConnection connection = EstablishConnection(testPeer: testPeer, setClientId: false);
// Tell the test peer to close the connection when executing its last handler
- testPeer.RemotelyCloseConnection(expectCloseResponse: true);
+ testPeer.RemotelyCloseConnection(expectCloseResponse: true, errorCondition: ConnectionError.CONNECTION_FORCED, errorMessage: "buba");
connection.ExceptionListener += exception => done.Set();
diff --git a/test/Apache-NMS-AMQP-Test/Integration/ConsumerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/ConsumerIntegrationTest.cs
index 3eec377..d85e93b 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/ConsumerIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/ConsumerIntegrationTest.cs
@@ -83,7 +83,7 @@
// Create a consumer, then remotely end it afterwards.
testPeer.ExpectReceiverAttach();
testPeer.ExpectLinkFlow();
- testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, errorMessage: errorMessage);
+ testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, errorMessage: errorMessage, delayBeforeSend: 400);
IQueue queue = session.GetQueue("myQueue");
IMessageConsumer consumer = session.CreateConsumer(queue);
@@ -322,7 +322,7 @@
testPeer.ExpectReceiverAttach();
testPeer.ExpectLinkFlow();
testPeer.RunAfterLastHandler(() => { consumerReady.WaitOne(2000); });
- testPeer.RemotelyCloseConnection(expectCloseResponse: true);
+ testPeer.DropAfterLastMatcher(delay: 10);
IMessageConsumer consumer = session.CreateConsumer(queue);
consumerReady.Set();
@@ -375,7 +375,7 @@
testPeer.ExpectReceiverAttach();
testPeer.ExpectLinkFlow();
- testPeer.RemotelyCloseConnection(expectCloseResponse: true);
+ testPeer.RemotelyCloseConnection(expectCloseResponse: true, errorCondition: ConnectionError.CONNECTION_FORCED, errorMessage: "buba");
IMessageConsumer consumer = session.CreateConsumer(queue);
diff --git a/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs
index e4fb515..816dc84 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs
@@ -24,6 +24,7 @@
using Apache.NMS;
using Apache.NMS.AMQP;
using Moq;
+using NLog;
using NMS.AMQP.Test.TestAmqp;
using NMS.AMQP.Test.TestAmqp.BasicTypes;
using NUnit.Framework;
@@ -33,7 +34,9 @@
[TestFixture]
public class FailoverIntegrationTest : IntegrationTestFixture
{
- [Test, Timeout(20_000)]
+ private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
+
+ [Test, Timeout(20_000), Ignore("Ignore as we cannot detect connection disconnect on Linux.")]
public void TestFailoverHandlesDropThenRejectionCloseAfterConnect()
{
using (TestAmqpPeer originalPeer = new TestAmqpPeer())
@@ -47,6 +50,10 @@
var originalUri = CreatePeerUri(originalPeer);
var rejectingUri = CreatePeerUri(rejectingPeer);
var finalUri = CreatePeerUri(finalPeer);
+
+ Logger.Info($"Original peer is at: {originalUri}");
+ Logger.Info($"Rejecting peer is at: {rejectingUri}");
+ Logger.Info($"Final peer is at: {finalUri}");
// Connect to the first
originalPeer.ExpectSaslAnonymous();
@@ -87,7 +94,7 @@
// Close the original peer and wait for things to shake out.
originalPeer.Close(sendClose: true);
- rejectingPeer.WaitForAllMatchersToComplete(1000);
+ rejectingPeer.WaitForAllMatchersToComplete(2000);
Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
DateTime end = DateTime.UtcNow;
@@ -305,7 +312,7 @@
testPeer.ExpectSaslAnonymous();
testPeer.ExpectOpen();
testPeer.ExpectBegin();
- testPeer.DropAfterLastMatcher();
+ testPeer.DropAfterLastMatcher(delay: 10);
NmsConnection connection = EstablishAnonymousConnection("nms.requestTimeout=1000&failover.reconnectDelay=2000&failover.maxReconnectAttempts=60", testPeer);
@@ -958,9 +965,9 @@
using (TestAmqpPeer originalPeer = new TestAmqpPeer())
using (TestAmqpPeer finalPeer = new TestAmqpPeer())
{
+ ManualResetEvent consumerReady = new ManualResetEvent(false);
ManualResetEvent originalConnected = new ManualResetEvent(false);
ManualResetEvent finalConnected = new ManualResetEvent(false);
- ManualResetEvent exceptionThrown = new ManualResetEvent(false);
// Connect to the first peer
originalPeer.ExpectSaslAnonymous();
@@ -969,7 +976,6 @@
originalPeer.ExpectBegin();
NmsConnection connection = EstablishAnonymousConnection(originalPeer, finalPeer);
- connection.ExceptionListener += exception => { exceptionThrown.Set(); };
Mock<INmsConnectionListener> connectionListener = new Mock<INmsConnectionListener>();
@@ -989,6 +995,7 @@
originalPeer.ExpectReceiverAttach();
originalPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), 1);
+ originalPeer.RunAfterLastHandler(() => consumerReady.WaitOne(TimeSpan.FromSeconds(2)));
originalPeer.DropAfterLastMatcher();
// Post Failover Expectations of FinalPeer
@@ -1003,17 +1010,21 @@
ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
IQueue queue = session.GetQueue("myQueue");
IMessageConsumer messageConsumer = session.CreateConsumer(queue);
- int msgReceivedCount = 0;
+ CountdownEvent msgReceivedLatch = new CountdownEvent(2);
messageConsumer.Listener += message =>
{
- finalConnected.WaitOne(TimeSpan.FromSeconds(5));
- msgReceivedCount++;
+ if (msgReceivedLatch.CurrentCount == 2)
+ {
+ consumerReady.Set();
+ finalConnected.WaitOne(2000);
+ }
+
+ msgReceivedLatch.Signal();
};
finalPeer.WaitForAllMatchersToComplete(5000);
- Assert.AreEqual(2, msgReceivedCount);
- Assert.IsTrue(exceptionThrown.WaitOne(TimeSpan.FromSeconds(1)));
+ Assert.IsTrue(msgReceivedLatch.Wait(TimeSpan.FromSeconds(10)), $"Expected 2 messages, but got {2 - msgReceivedLatch.CurrentCount}");
}
}
diff --git a/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs b/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs
index cd52602..9a13133 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs
@@ -27,7 +27,7 @@
{
static IntegrationTestFixture()
{
- Tracer.Trace = new Logger(Logger.LogLevel.DEBUG);
+ Tracer.Trace = new NLogAdapter();
}
protected IConnection EstablishConnection(TestAmqpPeer testPeer, string optionsString = null, Symbol[] serverCapabilities = null, Fields serverProperties = null, bool setClientId = true)
diff --git a/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs
index 11d7d66..b2b2d7e 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs
@@ -543,7 +543,7 @@
// Create a producer, then remotely end it afterwards.
testPeer.ExpectSenderAttach();
- testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, breadCrumb);
+ testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, breadCrumb, delayBeforeSend: 10);
IQueue destination = session.GetQueue("myQueue");
IMessageProducer producer = session.CreateProducer(destination);
diff --git a/test/Apache-NMS-AMQP-Test/Message/Facade/NmsTestMessageFacade.cs b/test/Apache-NMS-AMQP-Test/Message/Facade/NmsTestMessageFacade.cs
index b1d017d..cc3a697 100644
--- a/test/Apache-NMS-AMQP-Test/Message/Facade/NmsTestMessageFacade.cs
+++ b/test/Apache-NMS-AMQP-Test/Message/Facade/NmsTestMessageFacade.cs
@@ -75,7 +75,7 @@
public string GroupId { get; set; }
public uint GroupSequence { get; set; }
public DateTime? Expiration { get; set; }
- public sbyte JmsMsgType { get; }
+ public sbyte? JmsMsgType { get; }
public bool IsPersistent { get; set; }
public INmsMessageFacade Copy()
diff --git a/test/Apache-NMS-AMQP-Test/NLog.config b/test/Apache-NMS-AMQP-Test/NLog.config
new file mode 100644
index 0000000..71b7c9c
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/NLog.config
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="utf-8"?>
+
+<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+
+ <targets>
+ <target name="logconsole" xsi:type="Console" layout="${time} | ${level} | ${callsite:includeNamespace=false:methodName=false} | ${message}" />
+ </targets>
+
+ <rules>
+ <logger name="*" minlevel="Debug" writeTo="logconsole" />
+ </rules>
+</nlog>
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpCodecTest.cs b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpCodecTest.cs
index ffeb404..5677366 100644
--- a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpCodecTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpCodecTest.cs
@@ -458,7 +458,6 @@
DoTestNMSMessageEncodingAddsProperMessageAnnotations(MessageSupport.JMS_TYPE_MAP, null, MessageSupport.JMS_DEST_TYPE_TEMP_QUEUE);
DoTestNMSMessageEncodingAddsProperMessageAnnotations(MessageSupport.JMS_TYPE_TXT, MessageSupport.JMS_DEST_TYPE_TOPIC, MessageSupport.JMS_DEST_TYPE_TEMP_QUEUE);
DoTestNMSMessageEncodingAddsProperMessageAnnotations(MessageSupport.JMS_TYPE_STRM, MessageSupport.JMS_DEST_TYPE_QUEUE, MessageSupport.JMS_DEST_TYPE_TEMP_QUEUE);
- DoTestNMSMessageEncodingAddsProperMessageAnnotations(MessageSupport.JMS_TYPE_OBJ, MessageSupport.JMS_DEST_TYPE_QUEUE, MessageSupport.JMS_DEST_TYPE_TEMP_QUEUE);
}
private void DoTestNMSMessageEncodingAddsProperMessageAnnotations(sbyte msgType, byte? toType, byte? replyToType)
diff --git a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpMessageFactoryTest.cs b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpMessageFactoryTest.cs
index 450963a..562048f 100644
--- a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpMessageFactoryTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpMessageFactoryTest.cs
@@ -121,7 +121,7 @@
Assert.IsInstanceOf<NmsObjectMessage>(message);
Assert.IsInstanceOf<AmqpNmsObjectMessageFacade>(facade);
- Assert.AreEqual(MessageSupport.JMS_TYPE_OBJ, facade.JmsMsgType);
+ Assert.IsNull(facade.JmsMsgType);
Assert.IsNull(((AmqpNmsObjectMessageFacade) facade).Body);
}
@@ -137,7 +137,7 @@
Assert.IsInstanceOf<NmsObjectMessage>(message);
Assert.IsInstanceOf<AmqpNmsObjectMessageFacade>(facade);
- Assert.AreEqual(MessageSupport.JMS_TYPE_OBJ, facade.JmsMsgType);
+ Assert.IsNull(facade.JmsMsgType);
AmqpNmsObjectMessageFacade objectMessageFacade = (AmqpNmsObjectMessageFacade) facade;
diff --git a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpNmsObjectMessageFacadeTest.cs b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpNmsObjectMessageFacadeTest.cs
index 015b64b..29a077f 100644
--- a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpNmsObjectMessageFacadeTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpNmsObjectMessageFacadeTest.cs
@@ -39,7 +39,7 @@
AmqpNmsObjectMessageFacade amqpObjectMessageFacade = CreateNewObjectMessageFacade(false);
Assert.Null(amqpObjectMessageFacade.MessageAnnotations);
- Assert.AreEqual(MessageSupport.JMS_TYPE_OBJ, amqpObjectMessageFacade.JmsMsgType);
+ Assert.IsNull(amqpObjectMessageFacade.JmsMsgType);
}
[Test]
diff --git a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpProviderFactoryTest.cs b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpProviderFactoryTest.cs
index 3bf5383..ea0de01 100644
--- a/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpProviderFactoryTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Provider/Amqp/AmqpProviderFactoryTest.cs
@@ -16,6 +16,7 @@
*/
using System;
+using Amqp;
using Apache.NMS.AMQP.Provider;
using Apache.NMS.AMQP.Provider.Amqp;
using NUnit.Framework;
@@ -26,7 +27,7 @@
public class AmqpProviderFactoryTest
{
private uint customMaxHandle = 2048;
-
+
[Test]
public void TestCreateAmqpProvider()
{
@@ -38,21 +39,31 @@
public void TestCreateWithDefaultOptions()
{
AmqpProvider provider = ProviderFactory.Create(new Uri("amqp://localhost:5672")) as AmqpProvider;
-
+
Assert.IsNotNull(provider);
Assert.AreEqual(AmqpProvider.DEFAULT_MAX_HANDLE, provider.MaxHandle);
+ Assert.IsFalse(provider.TraceFrames);
}
[Test]
public void TestCreateWithCustomOptions()
{
- Uri uri = new Uri("amqp://localhost:5672" + "?" +
- "amqp.maxHandle=" + customMaxHandle);
-
+ Uri uri = new Uri("amqp://localhost:5672" +
+ "?amqp.maxHandle=" + customMaxHandle +
+ "&amqp.traceFrames=true");
+
AmqpProvider provider = ProviderFactory.Create(uri) as AmqpProvider;
-
+
Assert.IsNotNull(provider);
Assert.AreEqual(customMaxHandle, provider.MaxHandle);
+ Assert.IsTrue(provider.TraceFrames);
+ }
+
+ [TearDown]
+ public void TearDown()
+ {
+ // clean up trace level as it may interfere with other tests
+ Trace.TraceLevel = 0;
}
}
}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/BasicTypes/ConnectionError.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/BasicTypes/ConnectionError.cs
new file mode 100644
index 0000000..6889a62
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/BasicTypes/ConnectionError.cs
@@ -0,0 +1,9 @@
+using Amqp.Types;
+
+namespace NMS.AMQP.Test.TestAmqp.BasicTypes
+{
+ public static class ConnectionError
+ {
+ public static readonly Symbol CONNECTION_FORCED = new Symbol("amqp:connection:forced");
+ }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/Logger.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/Logger.cs
deleted file mode 100644
index ce24b3d..0000000
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/Logger.cs
+++ /dev/null
@@ -1,89 +0,0 @@
-using System;
-using Apache.NMS;
-
-namespace NMS.AMQP.Test.TestAmqp
-{
- class Logger : ITrace
- {
- public enum LogLevel
- {
- OFF = -1,
- FATAL,
- ERROR,
- WARN,
- INFO,
- DEBUG
- }
-
- private LogLevel lv;
-
- public void LogException(Exception ex)
- {
- this.Warn("Exception: " + ex.Message);
- }
-
- public Logger() : this(LogLevel.WARN)
- {
- }
-
- public Logger(LogLevel lvl)
- {
- lv = lvl;
- }
-
- public bool IsDebugEnabled
- {
- get { return lv >= LogLevel.DEBUG; }
- }
-
- public bool IsErrorEnabled
- {
- get { return lv >= LogLevel.ERROR; }
- }
-
- public bool IsFatalEnabled
- {
- get { return lv >= LogLevel.FATAL; }
- }
-
- public bool IsInfoEnabled
- {
- get { return lv >= LogLevel.INFO; }
- }
-
- public bool IsWarnEnabled
- {
- get { return lv >= LogLevel.WARN; }
- }
-
- public void Debug(string message)
- {
- if (IsDebugEnabled)
- Console.WriteLine("Debug: {0}", message);
- }
-
- public void Error(string message)
- {
- if (IsErrorEnabled)
- Console.WriteLine("Error: {0}", message);
- }
-
- public void Fatal(string message)
- {
- if (IsFatalEnabled)
- Console.WriteLine("Fatal: {0}", message);
- }
-
- public void Info(string message)
- {
- if (IsInfoEnabled)
- Console.WriteLine("Info: {0}", message);
- }
-
- public void Warn(string message)
- {
- if (IsWarnEnabled)
- Console.WriteLine("Warn: {0}", message);
- }
- }
-}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs
new file mode 100644
index 0000000..326be86
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs
@@ -0,0 +1,30 @@
+using Apache.NMS;
+using NLog;
+
+namespace NMS.AMQP.Test.TestAmqp
+{
+ class NLogAdapter : ITrace
+ {
+ private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
+
+ public bool IsDebugEnabled => Logger.IsDebugEnabled;
+
+ public bool IsErrorEnabled => Logger.IsErrorEnabled;
+
+ public bool IsFatalEnabled => Logger.IsFatalEnabled;
+
+ public bool IsInfoEnabled => Logger.IsInfoEnabled;
+
+ public bool IsWarnEnabled => Logger.IsWarnEnabled;
+
+ public void Debug(string message) => Logger.Debug(message);
+
+ public void Error(string message) => Logger.Error(message);
+
+ public void Fatal(string message) => Logger.Fatal(message);
+
+ public void Info(string message) => Logger.Info(message);
+
+ public void Warn(string message) => Logger.Warn(message);
+ }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
index 9a519f2..5646215 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
@@ -28,6 +28,7 @@
using Amqp.Sasl;
using Amqp.Types;
using Apache.NMS.AMQP.Util;
+using NLog;
using NMS.AMQP.Test.TestAmqp.BasicTypes;
using NMS.AMQP.Test.TestAmqp.Matchers;
using NUnit.Framework;
@@ -36,6 +37,8 @@
{
public class TestAmqpPeer : IDisposable
{
+ private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
+
public static readonly string MESSAGE_NUMBER = "MessageNumber";
private static readonly Symbol ANONYMOUS = new Symbol("ANONYMOUS");
@@ -93,6 +96,8 @@
public bool OnFrame(Stream stream, ushort channel, DescribedList command, Amqp.Message message)
{
+ Logger.Debug($"Received frame, descriptor={command.GetType().Name} on port: {ServerPort}");
+
var matcher = GetFirstMatcher();
if (matcher != null)
{
@@ -327,10 +332,19 @@
message.ApplicationProperties[MESSAGE_NUMBER] = i;
}
+ if (message.Properties == null)
+ message.Properties = new Properties();
+
+ message.Properties.MessageId = $"ID:{i.ToString()}";
+
+ var messageId = message.Properties.MessageId;
+
ByteBuffer payload = message.Encode();
flowMatcher.WithOnComplete(context =>
{
+ Logger.Debug($"Sending message {messageId}");
+
var transfer = new Transfer()
{
DeliveryId = (uint) nextId,
@@ -340,7 +354,15 @@
Handle = context.Command.Handle,
};
- context.SendCommand(transfer, payload);
+ try
+ {
+ context.SendCommand(transfer, payload);
+ }
+ catch (Exception e)
+ {
+ Logger.Error($"Sending message {messageId} failed.");
+ throw;
+ }
});
}
@@ -837,6 +859,8 @@
public void Close(bool sendClose = false)
{
+ Logger.Info($"Closing {nameof(TestAmqpPeer)}: {this.ServerPort}");
+
if (sendClose)
{
var close = new Close();
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs
index f0341cd..21f0687 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs
@@ -23,12 +23,15 @@
using Amqp;
using Amqp.Framing;
using Amqp.Types;
+using NLog.Fluent;
using NMS.AMQP.Test.TestAmqp.BasicTypes;
namespace NMS.AMQP.Test.TestAmqp
{
public class TestAmqpPeerRunner
{
+ private static readonly NLog.Logger Logger = NLog.LogManager.GetCurrentClassLogger();
+
private readonly TestAmqpPeer testAmqpPeer;
private readonly IPEndPoint ip;
private Socket socket;
@@ -123,8 +126,9 @@
}
}
}
- catch
+ catch(Exception e)
{
+ Logger.Info(e);
stream.Dispose();
}
}
diff --git a/test/Apache-NMS-AMQP-Test/Transport/TransportContextFactoryTest.cs b/test/Apache-NMS-AMQP-Test/Transport/TransportContextFactoryTest.cs
index 8669370..5e69210 100644
--- a/test/Apache-NMS-AMQP-Test/Transport/TransportContextFactoryTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Transport/TransportContextFactoryTest.cs
@@ -29,6 +29,8 @@
private int customReceiveTimeout = 1000;
private int customSendBufferSize = 32 * 1024;
private int customSendTimeout = 2000;
+ private int customTcpKeepAliveTime = 2500;
+ private int customTcpKeepAliveInterval = 3000;
[Test]
public void TestCreateWithDefaultOptions()
@@ -54,6 +56,8 @@
"transport.receiveTimeout=" + customReceiveTimeout + "&" +
"transport.sendBufferSize=" + customSendBufferSize + "&" +
"transport.sendTimeout=" + customSendTimeout + "&" +
+ "transport.tcpKeepAliveTime=" + customTcpKeepAliveTime + "&" +
+ "transport.tcpKeepAliveInterval=" + customTcpKeepAliveInterval + "&" +
"transport.tcpNoDelay=" + customTcpNoDelay);
ITransportContext transportContext = TransportContextFactory.CreateTransportContext(uri);
@@ -64,6 +68,8 @@
Assert.AreEqual(customReceiveTimeout, transportContext.ReceiveTimeout);
Assert.AreEqual(customSendBufferSize, transportContext.SendBufferSize);
Assert.AreEqual(customSendTimeout, transportContext.SendTimeout);
+ Assert.AreEqual(customTcpKeepAliveTime, transportContext.TcpKeepAliveTime);
+ Assert.AreEqual(customTcpKeepAliveInterval, transportContext.TcpKeepAliveInterval);
Assert.AreEqual(customTcpNoDelay, transportContext.TcpNoDelay);
}
