Merge pull request #17 from syncromatics/hotfix/amqp-publish-overload

(+semver: fix) AMQP publish with routing key

amqp/exchangePublisher.go

@@ -52,3 +52,21 @@ func (p *ExchangePublisher) Publish(exchangeName string, headers map[string]stri
 
 	return nil
 }
+
+// PublishWithRoutingKey publishes a message to the given exchange, with a routing key to specify the queue
+func (p *ExchangePublisher) PublishWithRoutingKey(exchangeName string, routingKey string, body []byte) error {
+	channel, err := p.connection.Channel()
+	if err != nil {
+		return errors.Wrap(err, "failed to open channel to broker")
+	}
+	defer channel.Close()
+
+	err = channel.Publish(exchangeName, routingKey, false, false, amqp.Publishing{
+		Body: body,
+	})
+	if err != nil {
+		return errors.Wrap(err, "failed to publish message")
+	}
+
+	return nil
+}

amqp/exchangePublisher_test.go

@@ -88,3 +88,44 @@ func Test_Publish_Successful(t *testing.T) {
 	}
 
 }
+
+func Test_PublishWithRoutingKey_MatchSuccessful(t *testing.T) {
+	// Arrange
+	conn, err := amqp.Dial(amqpURL)
+	assert.Nil(t, err)
+	defer conn.Close()
+
+	channel, err := conn.Channel()
+	assert.Nil(t, err)
+	defer channel.Close()
+
+	testQueueName := fmt.Sprintf("test.%v", uuid.NewString())
+	_, err = channel.QueueDeclare(testQueueName, false, true, true, false, nil)
+	assert.Nil(t, err)
+
+	err = channel.QueueBind(testQueueName, "prefix.123.#", EXCHANGE_NAME, false, nil)
+	assert.Nil(t, err)
+
+	actualMessages, err := channel.Consume(testQueueName, uuid.NewString(), true, true, false, false, nil)
+	assert.Nil(t, err)
+
+	publisher := sut.NewExchangePublisher(amqpURL)
+
+	err = publisher.EnsurePublisherIsReady()
+	assert.Nil(t, err)
+
+	expectedBody := []byte(`{"VehicleId":1}`)
+
+	// Act
+	err = publisher.PublishWithRoutingKey(EXCHANGE_NAME, "prefix.123.otherstuff.789", expectedBody)
+
+	// Assert
+	assert.Nil(t, err)
+
+	select {
+	case actual := <-actualMessages:
+		assert.Equal(t, expectedBody, actual.Body)
+	case <-time.After(3 * time.Second):
+		assert.Fail(t, "expected to receive message within a timely manner")
+	}
+}