Repository: Azure/azure-service-bus-go
Pull request #75 (2018-12-04 23:33:58)

add dead letter recievers for entities
#41 


Patch:
diff --git a/changelog.md b/changelog.md
index 1d8418d..d10fbf5 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,4 +1,10 @@
 # Change Log
 
+## `head`
+- Refactor disposition handler so that errors can be handled in handlers
+- Add dead letter queues for entities
+- Fix connection leaks when using multiple calls to Receive
+- Ensure senders wait for message disposition before returning
+
 ## `v0.1.0`
 - initial tag for Service Bus which includes Queues, Topics and Subscriptions using AMQP
\ No newline at end of file
diff --git a/deadletter.go b/deadletter.go
new file mode 100644
index 0000000..61db62c
--- /dev/null
+++ b/deadletter.go
@@ -0,0 +1,161 @@
+package servicebus
+
+import (
+	"context"
+	"sync"
+
+	"github.com/Azure/azure-amqp-common-go/log"
+)
+
+type (
+	// Closer provides the ability to close an entity
+	Closer interface {
+		Close(ctx context.Context) error
+	}
+
+	// ReceiveOner provides the ability to receive and handle events
+	ReceiveOner interface {
+		Closer
+		ReceiveOne(ctx context.Context, handler Handler) error
+	}
+
+	// DeadLetterBuilder provides the ability to create a new receiver addressed to a given entity's dead letter queue.
+	DeadLetterBuilder interface {
+		NewDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error)
+	}
+
+	// TransferDeadLetterBuilder provides the ability to create a new receiver addressed to a given entity's transfer
+	// dead letter queue.
+	TransferDeadLetterBuilder interface {
+		NewTransferDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error)
+	}
+
+	// DeadLetter represents a dead letter queue in Azure Service Bus.
+	//
+	// Azure Service Bus queues, topics and subscriptions provide a secondary sub-queue, called a dead-letter queue
+	// (DLQ). The dead-letter queue does not need to be explicitly created and cannot be deleted or otherwise managed
+	// independent of the main entity.
+	//
+	// The purpose of the dead-letter queue is to hold messages that cannot be delivered to any receiver, or messages
+	// that could not be processed. Messages can then be removed from the DLQ and inspected. An application might, with
+	// help of an operator, correct issues and resubmit the message, log the fact that there was an error, and take
+	// corrective action.
+	//
+	// From an API and protocol perspective, the DLQ is mostly similar to any other queue, except that messages can only
+	// be submitted via the dead-letter operation of the parent entity. In addition, time-to-live is not observed, and
+	// you can't dead-letter a message from a DLQ. The dead-letter queue fully supports peek-lock delivery and
+	// transactional operations.
+	//
+	// Note that there is no automatic cleanup of the DLQ. Messages remain in the DLQ until you explicitly retrieve
+	// them from the DLQ and call Complete() on the dead-letter message.
+	DeadLetter struct {
+		builder  DeadLetterBuilder
+		rMu      sync.Mutex
+		receiver ReceiveOner
+	}
+
+	// TransferDeadLetter represents a transfer dead letter queue in Azure Service Bus.
+	//
+	// Messages will be sent to the transfer dead-letter queue under the following conditions:
+	//   - A message passes through more than 3 queues or topics that are chained together.
+	//   - The destination queue or topic is disabled or deleted.
+	//   - The destination queue or topic exceeds the maximum entity size.
+	TransferDeadLetter struct {
+		builder  TransferDeadLetterBuilder
+		rMu      sync.Mutex
+		receiver ReceiveOner
+	}
+)
+
+// NewDeadLetter constructs an instance of DeadLetter which represents a dead letter queue in Azure Service Bus
+func NewDeadLetter(builder DeadLetterBuilder) *DeadLetter {
+	return &DeadLetter{
+		builder: builder,
+	}
+}
+
+// ReceiveOne will receive one message from the dead letter queue
+func (dl *DeadLetter) ReceiveOne(ctx context.Context, handler Handler) error {
+	if err := dl.ensureReceiver(ctx); err != nil {
+		return err
+	}
+
+	return dl.receiver.ReceiveOne(ctx, handler)
+}
+
+// Close the underlying connection to Service Bus
+func (dl *DeadLetter) Close(ctx context.Context) error {
+	dl.rMu.Lock()
+	defer dl.rMu.Unlock()
+
+	if dl.receiver == nil {
+		return nil
+	}
+
+	if err := dl.receiver.Close(ctx); err != nil {
+		log.For(ctx).Error(err)
+		return err
+	}
+
+	return nil
+}
+
+func (dl *DeadLetter) ensureReceiver(ctx context.Context) error {
+	dl.rMu.Lock()
+	defer dl.rMu.Unlock()
+
+	r, err := dl.builder.NewDeadLetterReceiver(ctx)
+	if err != nil {
+		return err
+	}
+
+	dl.receiver = r
+	return nil
+}
+
+// NewTransferDeadLetter constructs an instance of DeadLetter which represents a transfer dead letter queue in
+// Azure Service Bus
+func NewTransferDeadLetter(builder TransferDeadLetterBuilder) *TransferDeadLetter {
+	return &TransferDeadLetter{
+		builder: builder,
+	}
+}
+
+// ReceiveOne will receive one message from the dead letter queue
+func (dl *TransferDeadLetter) ReceiveOne(ctx context.Context, handler Handler) error {
+	if err := dl.ensureReceiver(ctx); err != nil {
+		return err
+	}
+
+	return dl.receiver.ReceiveOne(ctx, handler)
+}
+
+// Close the underlying connection to Service Bus
+func (dl *TransferDeadLetter) Close(ctx context.Context) error {
+	dl.rMu.Lock()
+	defer dl.rMu.Unlock()
+
+	if dl.receiver == nil {
+		return nil
+	}
+
+	if err := dl.receiver.Close(ctx); err != nil {
+		log.For(ctx).Error(err)
+		return err
+	}
+
+	return nil
+}
+
+func (dl *TransferDeadLetter) ensureReceiver(ctx context.Context) error {
+	dl.rMu.Lock()
+	defer dl.rMu.Unlock()
+
+	r, err := dl.builder.NewTransferDeadLetterReceiver(ctx)
+	if err != nil {
+		return err
+	}
+
+	dl.receiver = r
+	return nil
+}
diff --git a/queue.go b/queue.go
index 4e7cf3a..09d66ca 100644
--- a/queue.go
+++ b/queue.go
@@ -27,6 +27,7 @@ import (
 	"encoding/xml"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -104,6 +105,13 @@ const (
 	PeekLockMode ReceiveMode = 0
 	// ReceiveAndDeleteMode causes a Receiver to pop messages off of the queue without waiting for DispositionAction
 	ReceiveAndDeleteMode ReceiveMode = 1
+
+	// DeadLetterQueueName is the name of the dead letter queue to be appended to the entity path
+	DeadLetterQueueName = "$DeadLetterQueue"
+
+	// TransferDeadLetterQueueName is the name of the transfer dead letter queue which is appended to the entity name to
+	// build the full address of the transfer dead letter queue.
+	TransferDeadLetterQueueName = "$Transfer/" + DeadLetterQueueName
 )
 
 // QueueWithReceiveAndDelete configures a queue to pop and delete messages off of the queue upon receiving the message.
@@ -366,15 +374,76 @@ func (q *Queue) NewSession(sessionID *string) *QueueSession {
 
 // NewReceiver will create a new Receiver for receiving messages off of a queue
 func (q *Queue) NewReceiver(ctx context.Context, opts ...ReceiverOption) (*Receiver, error) {
+	span, ctx := q.startSpanFromContext(ctx, "sb.Queue.NewReceiver")
+	defer span.Finish()
+
 	opts = append(opts, ReceiverWithReceiveMode(q.receiveMode))
 	return q.namespace.NewReceiver(ctx, q.Name, opts...)
 }
 
 // NewSender will create a new Sender for sending messages to the queue
 func (q *Queue) NewSender(ctx context.Context, opts ...SenderOption) (*Sender, error) {
+	span, ctx := q.startSpanFromContext(ctx, "sb.Queue.NewSender")
+	defer span.Finish()
+
 	return q.namespace.NewSender(ctx, q.Name)
 }
 
+// NewDeadLetter creates an entity that represents the dead letter sub queue of the queue
+//
+// Azure Service Bus queues and topic subscriptions provide a secondary sub-queue, called a dead-letter queue
+// (DLQ). The dead-letter queue does not need to be explicitly created and cannot be deleted or otherwise managed
+// independent of the main entity.
+//
+// The purpose of the dead-letter queue is to hold messages that cannot be delivered to any receiver, or messages
+// that could not be processed. Messages can then be removed from the DLQ and inspected. An application might, with
+// help of an operator, correct issues and resubmit the message, log the fact that there was an error, and take
+// corrective action.
+//
+// From an API and protocol perspective, the DLQ is mostly similar to any other queue, except that messages can only
+// be submitted via the dead-letter operation of the parent entity. In addition, time-to-live is not observed, and
+// you can't dead-letter a message from a DLQ. The dead-letter queue fully supports peek-lock delivery and
+// transactional operations.
+//
+// Note that there is no automatic cleanup of the DLQ. Messages remain in the DLQ until you explicitly retrieve
+// them from the DLQ and call Complete() on the dead-letter message.
+func (q *Queue) NewDeadLetter() *DeadLetter {
+	return NewDeadLetter(q)
+}
+
+// NewDeadLetterReceiver builds a receiver for the Queue's dead letter queue
+func (q *Queue) NewDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	span, ctx := q.startSpanFromContext(ctx, "sb.Queue.NewDeadLetterReceiver")
+	defer span.Finish()
+
+	deadLetterEntityPath := strings.Join([]string{q.Name, DeadLetterQueueName}, "/")
+	return q.namespace.NewReceiver(ctx, deadLetterEntityPath, opts...)
+}
+
+// NewTransferDeadLetter creates an entity that represents the transfer dead letter sub queue of the queue
+//
+// Messages will be sent to the transfer dead-letter queue under the following conditions:
+//   - A message passes through more than 3 queues or topics that are chained together.
+//   - The destination queue or topic is disabled or deleted.
+//   - The destination queue or topic exceeds the maximum entity size.
+func (q *Queue) NewTransferDeadLetter() *TransferDeadLetter {
+	return NewTransferDeadLetter(q)
+}
+
+// NewTransferDeadLetterReceiver builds a receiver for the Queue's transfer dead letter queue
+//
+// Messages will be sent to the transfer dead-letter queue under the following conditions:
+//   - A message passes through more than 3 queues or topics that are chained together.
+//   - The destination queue or topic is disabled or deleted.
+//   - The destination queue or topic exceeds the maximum entity size.
+func (q *Queue) NewTransferDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	span, ctx := q.startSpanFromContext(ctx, "sb.Queue.NewTransferDeadLetterReceiver")
+	defer span.Finish()
+
+	transferDeadLetterEntityPath := strings.Join([]string{q.Name, TransferDeadLetterQueueName}, "/")
+	return q.namespace.NewReceiver(ctx, transferDeadLetterEntityPath, opts...)
+}
+
 // Close the underlying connection to Service Bus
 func (q *Queue) Close(ctx context.Context) error {
 	span, ctx := q.startSpanFromContext(ctx, "sb.Queue.Close")
diff --git a/subscription.go b/subscription.go
index dac4507..d0407d8 100644
--- a/subscription.go
+++ b/subscription.go
@@ -25,6 +25,7 @@ package servicebus
 import (
 	"context"
 	"encoding/xml"
+	"strings"
 	"sync"
 
 	"github.com/Azure/azure-amqp-common-go/log"
@@ -178,10 +179,68 @@ func (s *Subscription) NewSession(sessionID *string) *SubscriptionSession {
 
 // NewReceiver will create a new Receiver for receiving messages off of the queue
 func (s *Subscription) NewReceiver(ctx context.Context, opts ...ReceiverOption) (*Receiver, error) {
+	span, ctx := s.startSpanFromContext(ctx, "sb.Subscription.NewReceiver")
+	defer span.Finish()
+
 	opts = append(opts, ReceiverWithReceiveMode(s.receiveMode))
 	return s.namespace.NewReceiver(ctx, s.Topic.Name+"/Subscriptions/"+s.Name, opts...)
 }
 
+// NewDeadLetter creates an entity that represents the dead letter sub queue of the queue
+//
+// Azure Service Bus queues and topic subscriptions provide a secondary sub-queue, called a dead-letter queue
+// (DLQ). The dead-letter queue does not need to be explicitly created and cannot be deleted or otherwise managed
+// independent of the main entity.
+//
+// The purpose of the dead-letter queue is to hold messages that cannot be delivered to any receiver, or messages
+// that could not be processed. Messages can then be removed from the DLQ and inspected. An application might, with
+// help of an operator, correct issues and resubmit the message, log the fact that there was an error, and take
+// corrective action.
+//
+// From an API and protocol perspective, the DLQ is mostly similar to any other queue, except that messages can only
+// be submitted via the dead-letter operation of the parent entity. In addition, time-to-live is not observed, and
+// you can't dead-letter a message from a DLQ. The dead-letter queue fully supports peek-lock delivery and
+// transactional operations.
+//
+// Note that there is no automatic cleanup of the DLQ. Messages remain in the DLQ until you explicitly retrieve
+// them from the DLQ and call Complete() on the dead-letter message.
+func (s *Subscription) NewDeadLetter() *DeadLetter {
+	return NewDeadLetter(s)
+}
+
+// NewDeadLetterReceiver builds a receiver for the Subscriptions's dead letter queue
+func (s *Subscription) NewDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	span, ctx := s.startSpanFromContext(ctx, "sb.Subscription.NewDeadLetterReceiver")
+	defer span.Finish()
+
+	deadLetterEntityPath := strings.Join([]string{s.Topic.Name, "Subscriptions", s.Name, DeadLetterQueueName}, "/")
+	return s.namespace.NewReceiver(ctx, deadLetterEntityPath, opts...)
+}
+
+// NewTransferDeadLetter creates an entity that represents the transfer dead letter sub queue of the subscription
+//
+// Messages will be sent to the transfer dead-letter queue under the following conditions:
+//   - A message passes through more than 3 queues or topics that are chained together.
+//   - The destination queue or topic is disabled or deleted.
+//   - The destination queue or topic exceeds the maximum entity size.
+func (s *Subscription) NewTransferDeadLetter() *TransferDeadLetter {
+	return NewTransferDeadLetter(s)
+}
+
+// NewTransferDeadLetterReceiver builds a receiver for the Queue's transfer dead letter queue
+//
+// Messages will be sent to the transfer dead-letter queue under the following conditions:
+//   - A message passes through more than 3 queues or topics that are chained together.
+//   - The destination queue or topic is disabled or deleted.
+//   - The destination queue or topic exceeds the maximum entity size.
+func (s *Subscription) NewTransferDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	span, ctx := s.startSpanFromContext(ctx, "sb.Subscription.NewTransferDeadLetterReceiver")
+	defer span.Finish()
+
+	transferDeadLetterEntityPath := strings.Join([]string{s.Topic.Name, "Subscriptions", s.Name, TransferDeadLetterQueueName}, "/")
+	return s.namespace.NewReceiver(ctx, transferDeadLetterEntityPath, opts...)
+}
+
 // Close the underlying connection to Service Bus
 func (s *Subscription) Close(ctx context.Context) error {
 	if s.receiver != nil {
@@ -191,7 +250,7 @@ func (s *Subscription) Close(ctx context.Context) error {
 }
 
 func (s *Subscription) ensureReceiver(ctx context.Context, opts ...ReceiverOption) error {
-	span, ctx := s.startSpanFromContext(ctx, "sb.Queue.ensureReceiver")
+	span, ctx := s.startSpanFromContext(ctx, "sb.Subscription.ensureReceiver")
 	defer span.Finish()
 
 	s.receiverMu.Lock()
diff --git a/topic.go b/topic.go
index 03ef93b..e07a42f 100644
--- a/topic.go
+++ b/topic.go
@@ -25,6 +25,7 @@ package servicebus
 import (
 	"context"
 	"encoding/xml"
+	"strings"
 	"sync"
 
 	"github.com/Azure/azure-amqp-common-go/log"
@@ -132,6 +133,30 @@ func (t *Topic) Close(ctx context.Context) error {
 	return nil
 }
 
+// NewTransferDeadLetter creates an entity that represents the transfer dead letter sub queue of the topic
+//
+// Messages will be sent to the transfer dead-letter queue under the following conditions:
+//   - A message passes through more than 3 queues or topics that are chained together.
+//   - The destination queue or topic is disabled or deleted.
+//   - The destination queue or topic exceeds the maximum entity size.
+func (t *Topic) NewTransferDeadLetter() *TransferDeadLetter {
+	return NewTransferDeadLetter(t)
+}
+
+// NewTransferDeadLetterReceiver builds a receiver for the Queue's transfer dead letter queue
+//
+// Messages will be sent to the transfer dead-letter queue under the following conditions:
+//   - A message passes through more than 3 queues or topics that are chained together.
+//   - The destination queue or topic is disabled or deleted.
+//   - The destination queue or topic exceeds the maximum entity size.
+func (t *Topic) NewTransferDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	span, ctx := t.startSpanFromContext(ctx, "sb.Topic.NewTransferDeadLetterReceiver")
+	defer span.Finish()
+
+	transferDeadLetterEntityPath := strings.Join([]string{t.Name, TransferDeadLetterQueueName}, "/")
+	return t.namespace.NewReceiver(ctx, transferDeadLetterEntityPath, opts...)
+}
+
 func (t *Topic) ensureSender(ctx context.Context) error {
 	span, ctx := t.startSpanFromContext(ctx, "sb.Topic.ensureSender")
 	defer span.Finish()


Test patch:
diff --git a/deadletter_test.go b/deadletter_test.go
new file mode 100644
index 0000000..0bb4f8a
--- /dev/null
+++ b/deadletter_test.go
@@ -0,0 +1,137 @@
+package servicebus
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+type (
+	MockedDeadLetterBuilder struct {
+		mock.Mock
+		receiver *MockedReceiveOner
+	}
+
+	MockedReceiveOner struct {
+		mock.Mock
+	}
+
+	MockedTransferDeadLetterBuilder struct {
+		mock.Mock
+		receiver *MockedReceiveOner
+	}
+)
+
+func (ro *MockedReceiveOner) ReceiveOne(ctx context.Context, handler Handler) error {
+	args := ro.Called(ctx, handler)
+	return args.Error(0)
+}
+
+func (ro *MockedReceiveOner) Close(ctx context.Context) error {
+	args := ro.Called(ctx)
+	return args.Error(0)
+}
+
+func (m *MockedDeadLetterBuilder) NewDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	args := m.Called(ctx, opts)
+	return args.Get(0).(ReceiveOner), args.Error(1)
+}
+
+func (mt *MockedTransferDeadLetterBuilder) NewTransferDeadLetterReceiver(ctx context.Context, opts ...ReceiverOption) (ReceiveOner, error) {
+	args := mt.Called(ctx, opts)
+	return args.Get(0).(ReceiveOner), args.Error(1)
+}
+
+func TestTransferDeadLetter_ReceiveOne(t *testing.T) {
+	builder := setupDefaultDTLMock()
+	dl := NewTransferDeadLetter(builder)
+	err := dl.ReceiveOne(context.Background(), HandlerFunc(func(ctx context.Context, msg *Message) error {
+		return nil
+	}))
+	assert.NoError(t, err)
+	builder.AssertCalled(t, "NewTransferDeadLetterReceiver", context.Background(), mock.Anything)
+	builder.receiver.AssertCalled(t, "ReceiveOne", context.Background(), mock.Anything)
+}
+
+func TestTransferDeadLetter_Close(t *testing.T) {
+	builder := setupDefaultDTLMock()
+	dl := NewTransferDeadLetter(builder)
+	err := dl.ReceiveOne(context.Background(), HandlerFunc(func(ctx context.Context, msg *Message) error {
+		return nil
+	}))
+	assert.NoError(t, err)
+	err = dl.Close(context.Background())
+	assert.NoError(t, err)
+	builder.receiver.AssertCalled(t, "Close", context.Background())
+}
+
+func TestDeadLetter_ReceiveOne(t *testing.T) {
+	builder := setupDefaultDLMock()
+	dl := NewDeadLetter(builder)
+	err := dl.ReceiveOne(context.Background(), HandlerFunc(func(ctx context.Context, msg *Message) error {
+		return nil
+	}))
+	assert.NoError(t, err)
+	builder.AssertCalled(t, "NewDeadLetterReceiver", context.Background(), mock.Anything)
+	builder.receiver.AssertCalled(t, "ReceiveOne", context.Background(), mock.Anything)
+}
+
+func TestDeadLetter_Close(t *testing.T) {
+	builder := setupDefaultDLMock()
+	dl := NewDeadLetter(builder)
+	err := dl.ReceiveOne(context.Background(), HandlerFunc(func(ctx context.Context, msg *Message) error {
+		return nil
+	}))
+	assert.NoError(t, err)
+	err = dl.Close(context.Background())
+	assert.NoError(t, err)
+	builder.receiver.AssertCalled(t, "Close", context.Background())
+}
+
+func setupDefaultDLMock() *MockedDeadLetterBuilder {
+	receiver := new(MockedReceiveOner)
+	receiver.Mock.On(
+		"ReceiveOne",
+		context.Background(),
+		mock.AnythingOfType("servicebus.HandlerFunc"),
+	).Return(nil)
+	receiver.Mock.On(
+		"Close",
+		context.Background(),
+	).Return(nil)
+
+	builder := &MockedDeadLetterBuilder{
+		receiver: receiver,
+	}
+	builder.Mock.On(
+		"NewDeadLetterReceiver",
+		context.Background(),
+		mock.AnythingOfType("[]servicebus.ReceiverOption"),
+	).Return(receiver, nil)
+	return builder
+}
+
+func setupDefaultDTLMock() *MockedTransferDeadLetterBuilder {
+	receiver := new(MockedReceiveOner)
+	receiver.Mock.On(
+		"ReceiveOne",
+		context.Background(),
+		mock.AnythingOfType("servicebus.HandlerFunc"),
+	).Return(nil)
+	receiver.Mock.On(
+		"Close",
+		context.Background(),
+	).Return(nil)
+
+	builder := &MockedTransferDeadLetterBuilder{
+		receiver: receiver,
+	}
+	builder.Mock.On(
+		"NewTransferDeadLetterReceiver",
+		context.Background(),
+		mock.AnythingOfType("[]servicebus.ReceiverOption"),
+	).Return(receiver, nil)
+	return builder
+}
diff --git a/queue_test.go b/queue_test.go
index 0c6dbaf..d448a5b 100644
--- a/queue_test.go
+++ b/queue_test.go
@@ -27,14 +27,17 @@ import (
 	"encoding/xml"
 	"fmt"
 	"math/rand"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/servicebus/mgmt/2015-08-01/servicebus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	"github.com/Azure/azure-service-bus-go/atom"
 	"github.com/Azure/azure-service-bus-go/internal/test"
-	"github.com/stretchr/testify/assert"
 )
 
 const (
@@ -404,38 +407,18 @@ func buildQueue(ctx context.Context, t *testing.T, qm *QueueManager, name string
 
 func (suite *serviceBusSuite) TestQueueClient() {
 	tests := map[string]func(context.Context, *testing.T, *Queue){
-		"SimpleSend":         testQueueSend,
-		"DuplicateDetection": testDuplicateDetection,
-		"MessageProperties":  testMessageProperties,
-		"Retry":              testRequeueOnFail,
+		"SimpleSend_NoZeroCheck": testQueueSend,
+		"DuplicateDetection":     testDuplicateDetection,
+		"MessageProperties":      testMessageProperties,
+		"Retry":                  testRequeueOnFail,
 	}
 
-	ns := suite.getNewSasInstance()
-	for name, testFunc := range tests {
-		setupTestTeardown := func(t *testing.T) {
-			queueName := suite.randEntityName()
-
-			ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
-			defer cancel()
-
-			window := time.Duration(30 * time.Second)
-			cleanup := makeQueue(ctx, t, ns, queueName,
-				QueueEntityWithPartitioning(),
-				QueueEntityWithDuplicateDetection(&window))
-			q, err := ns.NewQueue(queueName)
-			suite.NoError(err)
-			defer func() {
-				q.Close(ctx)
-				cleanup()
-			}()
-			testFunc(ctx, t, q)
-			if !t.Failed() && name != "SimpleSend" {
-				checkZeroQueueMessages(ctx, t, ns, queueName)
-			}
-		}
-
-		suite.T().Run(name, setupTestTeardown)
+	window := time.Duration(30 * time.Second)
+	mgmtOpts := []QueueManagementOption{
+		QueueEntityWithPartitioning(),
+		QueueEntityWithDuplicateDetection(&window),
 	}
+	suite.queueMessageTestWithMgmtOptions(tests, mgmtOpts...)
 }
 
 func testRequeueOnFail(ctx context.Context, t *testing.T, q *Queue) {
@@ -543,27 +526,7 @@ func (suite *serviceBusSuite) TestQueueWithReceiveAndDelete() {
 	tests := map[string]func(context.Context, *testing.T, *Queue){
 		"SimpleSendAndReceive": testQueueSendAndReceiveWithReceiveAndDelete,
 	}
-
-	ns := suite.getNewSasInstance()
-	for name, testFunc := range tests {
-		setupTestTeardown := func(t *testing.T) {
-			queueName := suite.randEntityName()
-			ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
-			defer cancel()
-			cleanup := makeQueue(ctx, t, ns, queueName)
-			q, err := ns.NewQueue(queueName, QueueWithReceiveAndDelete())
-			suite.NoError(err)
-			defer func() {
-				cleanup()
-			}()
-			testFunc(ctx, t, q)
-			q.Close(ctx)
-			if !t.Failed() {
-				checkZeroQueueMessages(ctx, t, ns, queueName)
-			}
-		}
-		suite.T().Run(name, setupTestTeardown)
-	}
+	suite.queueMessageTestWithQueueOptions(tests, QueueWithReceiveAndDelete())
 }
 
 func testQueueSendAndReceiveWithReceiveAndDelete(ctx context.Context, t *testing.T, queue *Queue) {
@@ -606,56 +569,103 @@ func testQueueSendAndReceiveWithReceiveAndDelete(ctx context.Context, t *testing
 
 func (suite *serviceBusSuite) TestIssue73QueueClient() {
 	tests := map[string]func(context.Context, *testing.T, *Queue){
-		"SimpleSend200": func(ctx context.Context, t *testing.T, queue *Queue) {
+		"SimpleSend200_NoZeroCheck": func(ctx context.Context, t *testing.T, queue *Queue) {
 			for i := 0; i < 200; i++ {
 				testQueueSend(ctx, t, queue)
 			}
+			checkMessageCount(ctx, t, queue.namespace, queue.Name, 200)
 		},
 	}
+	window := time.Duration(20 * time.Second)
+	ttl := time.Duration(14 * 24 * time.Hour)
+	suite.queueMessageTestWithMgmtOptions(
+		tests,
+		QueueEntityWithMessageTimeToLive(&ttl),
+		QueueEntityWithDuplicateDetection(&window),
+		QueueEntityWithMaxDeliveryCount(10),
+		QueueEntityWithMaxSizeInMegabytes(1*Megabytes),
+	)
+}
+
+func (suite *serviceBusSuite) TestQueue_NewSession() {
+	ns := suite.getNewSasInstance()
+	q, err := ns.NewQueue("foo")
+	suite.NoError(err)
+	sessionID := "123"
+	qs := NewQueueSession(q, &sessionID)
+	suite.Equal(sessionID, *qs.SessionID())
+}
+
+func (suite *serviceBusSuite) TestQueue_NewDeadLetter() {
+	tests := map[string]func(context.Context, *testing.T, *Queue){
+		"ReceiveOneFromDeadLetter": testReceiveOneFromDeadLetter,
+	}
+	suite.queueMessageTestWithMgmtOptions(tests, QueueEntityWithMaxDeliveryCount(10))
+}
+
+func testReceiveOneFromDeadLetter(ctx context.Context, t *testing.T, q *Queue) {
+	qdl := q.NewDeadLetter()
+	require.NotNil(t, qdl)
+	require.NoError(t, q.Send(ctx, NewMessageFromString("foo")))
+
+	// abandon multiple 10 times until the message is dead lettered
+	for i := 0; i < 10; i++ {
+		err := q.ReceiveOne(ctx, HandlerFunc(func(ctx context.Context, msg *Message) error {
+			return msg.Abandon(ctx)
+		}))
+		require.NoError(t, err)
+	}
+
+	dl := q.NewDeadLetter()
+	called := false
+	err := dl.ReceiveOne(ctx, HandlerFunc(func(ctx context.Context, msg *Message) error {
+		assert.Equal(t, "foo", string(msg.Data))
+		called = true
+		return msg.Complete(ctx)
+	}))
+	assert.True(t, called)
+	assert.NoError(t, err)
+}
+
+func (suite *serviceBusSuite) queueMessageTestWithQueueOptions(
+	tests map[string]func(context.Context, *testing.T, *Queue),
+	queueOpts ...QueueOption) {
+	suite.queueMessageTest(tests, queueOpts, []QueueManagementOption{})
+}
+
+func (suite *serviceBusSuite) queueMessageTestWithMgmtOptions(
+	tests map[string]func(context.Context, *testing.T, *Queue),
+	mgmtOptions ...QueueManagementOption) {
+	suite.queueMessageTest(tests, []QueueOption{}, mgmtOptions)
+}
+
+func (suite *serviceBusSuite) queueMessageTest(
+	tests map[string]func(context.Context, *testing.T, *Queue),
+	queueOpts []QueueOption,
+	mgmtOptions []QueueManagementOption) {
 
 	ns := suite.getNewSasInstance()
 	for name, testFunc := range tests {
 		setupTestTeardown := func(t *testing.T) {
 			queueName := suite.randEntityName()
-
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
 			defer cancel()
-
-			window := time.Duration(20 * time.Second)
-			ttl := time.Duration(14 * 24 * time.Hour)
-			cleanup := makeQueue(ctx, t, ns, queueName,
-				QueueEntityWithMessageTimeToLive(&ttl),
-				QueueEntityWithDuplicateDetection(&window),
-				QueueEntityWithMaxDeliveryCount(10),
-				QueueEntityWithMaxSizeInMegabytes(1*Megabytes))
-			q, err := ns.NewQueue(queueName)
+			cleanup := makeQueue(ctx, t, ns, queueName, mgmtOptions...)
+			q, err := ns.NewQueue(queueName, queueOpts...)
 			suite.NoError(err)
 			defer func() {
-				q.Close(ctx)
 				cleanup()
 			}()
 			testFunc(ctx, t, q)
-			if !t.Failed() {
-
-			}
-			if !t.Failed() {
-				checkMessageCount(ctx, t, ns, queueName, 200)
+			suite.NoError(q.Close(ctx))
+			if !t.Failed() && !strings.HasSuffix(name, "NoZeroCheck") {
+				checkZeroQueueMessages(ctx, t, ns, queueName)
 			}
 		}
-
 		suite.T().Run(name, setupTestTeardown)
 	}
 }
 
-func (suite *serviceBusSuite) TestNewQueueSession() {
-	ns := suite.getNewSasInstance()
-	q, err := ns.NewQueue("foo")
-	suite.NoError(err)
-	sessionID := "123"
-	qs := NewQueueSession(q, &sessionID)
-	suite.Equal(sessionID, *qs.SessionID())
-}
-
 func makeQueue(ctx context.Context, t *testing.T, ns *Namespace, name string, opts ...QueueManagementOption) func() {
 	qm := ns.NewQueueManager()
 	entity, err := qm.Get(ctx, name)
diff --git a/subscription_test.go b/subscription_test.go
index 02fd7ec..3862db1 100644
--- a/subscription_test.go
+++ b/subscription_test.go
@@ -26,6 +26,7 @@ import (
 	"context"
 	"encoding/xml"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -237,39 +238,30 @@ func buildSubscription(ctx context.Context, t *testing.T, sm *SubscriptionManage
 	return s
 }
 
-func (suite *serviceBusSuite) TestSubscriptionClient() {
+func (suite *serviceBusSuite) TestSubscription_WithReceiveAndDelete() {
 	tests := map[string]func(context.Context, *testing.T, *Topic, *Subscription){
-		"SimpleReceive": testSubscriptionReceive,
-		"ReceiveOne":    testSubscriptionReceiveOne,
+		"ReceiveOne": testSubscriptionReceiveOneNoComplete,
 	}
 
-	ns := suite.getNewSasInstance()
-	for name, testFunc := range tests {
-		setupTestTeardown := func(t *testing.T) {
-			topicName := suite.randEntityName()
-			ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
-			defer cancel()
-
-			topicCleanup := makeTopic(ctx, t, ns, topicName)
-			topic, err := ns.NewTopic(topicName)
-			if suite.NoError(err) {
-				subName := suite.randEntityName()
-				subCleanup := makeSubscription(ctx, t, topic, subName)
-				subscription, err := topic.NewSubscription(subName)
+	suite.subscriptionMessageTestWithOptions(tests, SubscriptionWithReceiveAndDelete())
+}
 
-				if suite.NoError(err) {
-					defer subCleanup()
-					testFunc(ctx, t, topic, subscription)
-					if !t.Failed() {
-						checkZeroSubscriptionMessages(ctx, t, topic, subName)
-					}
-				}
-				defer topicCleanup()
-			}
-		}
+func testSubscriptionReceiveOneNoComplete(ctx context.Context, t *testing.T, topic *Topic, sub *Subscription) {
+	if assert.NoError(t, topic.Send(ctx, NewMessageFromString("hello!"))) {
+		err := sub.ReceiveOne(ctx, HandlerFunc(func(ctx context.Context, msg *Message) error {
+			return nil
+		}))
+		assert.NoError(t, err)
+	}
+}
 
-		suite.T().Run(name, setupTestTeardown)
+func (suite *serviceBusSuite) TestSubscriptionClient() {
+	tests := map[string]func(context.Context, *testing.T, *Topic, *Subscription){
+		"SimpleReceive": testSubscriptionReceive,
+		"ReceiveOne":    testSubscriptionReceiveOne,
 	}
+
+	suite.subscriptionMessageTestWithMgmtOptions(tests)
 }
 
 func testSubscriptionReceive(ctx context.Context, t *testing.T, topic *Topic, sub *Subscription) {
@@ -292,6 +284,35 @@ func testSubscriptionReceiveOne(ctx context.Context, t *testing.T, topic *Topic,
 	}
 }
 
+func (suite *serviceBusSuite) TestSubscription_NewDeadLetter() {
+	tests := map[string]func(context.Context, *testing.T, *Topic, *Subscription){
+		"ReceiveOneFromDeadLetter": testSubscriptionReceiveOneFromDeadLetter,
+	}
+
+	suite.subscriptionMessageTestWithOptions(tests)
+}
+
+func testSubscriptionReceiveOneFromDeadLetter(ctx context.Context, t *testing.T, topic *Topic, sub *Subscription) {
+	require.NoError(t, topic.Send(ctx, NewMessageFromString("foo")))
+
+	for i := 0; i < 10; i++ {
+		err := sub.ReceiveOne(ctx, HandlerFunc(func(ctx context.Context, msg *Message) error {
+			return msg.Abandon(ctx)
+		}))
+		require.NoError(t, err)
+	}
+
+	dl := sub.NewDeadLetter()
+	called := false
+	err := dl.ReceiveOne(ctx, HandlerFunc(func(ctx context.Context, msg *Message) error {
+		called = true
+		assert.Equal(t, "foo", string(msg.Data))
+		return msg.Complete(ctx)
+	}))
+	assert.True(t, called)
+	assert.NoError(t, err)
+}
+
 func (suite *serviceBusSuite) TestSubscriptionSessionClient() {
 	tests := map[string]func(context.Context, *testing.T, *TopicSession, *SubscriptionSession){
 		"ReceiveOne": testSubscriptionSessionReceiveOne,
@@ -317,7 +338,7 @@ func (suite *serviceBusSuite) TestSubscriptionSessionClient() {
 				ts := topic.NewSession(&sessionID)
 				ss := subscription.NewSession(&sessionID)
 				defer func() {
-					closeCtx, cancel := context.WithTimeout(ctx, 10 * time.Second)
+					closeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
 					defer cancel()
 					suite.NoError(ts.Close(closeCtx))
 					suite.NoError(ss.Close(closeCtx))
@@ -356,6 +377,44 @@ func testSubscriptionSessionReceiveOne(ctx context.Context, t *testing.T, topic
 	assert.Error(t, err, "context canceled")
 }
 
+func (suite *serviceBusSuite) subscriptionMessageTestWithOptions(tests map[string]func(context.Context, *testing.T, *Topic, *Subscription), opts ...SubscriptionOption) {
+	suite.subscriptionMessageTest(tests, opts, []SubscriptionManagementOption{})
+}
+
+func (suite *serviceBusSuite) subscriptionMessageTestWithMgmtOptions(tests map[string]func(context.Context, *testing.T, *Topic, *Subscription), mgmtOpts ...SubscriptionManagementOption) {
+	suite.subscriptionMessageTest(tests, []SubscriptionOption{}, mgmtOpts)
+}
+
+func (suite *serviceBusSuite) subscriptionMessageTest(tests map[string]func(context.Context, *testing.T, *Topic, *Subscription), subOpts []SubscriptionOption, mgmtOpts []SubscriptionManagementOption) {
+	ns := suite.getNewSasInstance()
+	for name, testFunc := range tests {
+		setupTestTeardown := func(t *testing.T) {
+			topicName := suite.randEntityName()
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+			defer cancel()
+
+			topicCleanup := makeTopic(ctx, t, ns, topicName)
+			topic, err := ns.NewTopic(topicName)
+			if suite.NoError(err) {
+				subName := suite.randEntityName()
+				subCleanup := makeSubscription(ctx, t, topic, subName, mgmtOpts...)
+				subscription, err := topic.NewSubscription(subName, subOpts...)
+
+				if suite.NoError(err) {
+					defer subCleanup()
+					testFunc(ctx, t, topic, subscription)
+					if !t.Failed() && !strings.HasSuffix(name, "NoZeroCheck") {
+						checkZeroSubscriptionMessages(ctx, t, topic, subName)
+					}
+				}
+				defer topicCleanup()
+			}
+		}
+
+		suite.T().Run(name, setupTestTeardown)
+	}
+}
+
 func makeSubscription(ctx context.Context, t *testing.T, topic *Topic, name string, opts ...SubscriptionManagementOption) func() {
 	sm := topic.NewSubscriptionManager()
 	entity, err := sm.Get(ctx, name)