Detect closed channel

messenger.go

@@ -90,6 +90,13 @@ func NewAMQPMessenger(c BrokerConfig, tlsConfig *tls.Config) (*AMQPMessenger, er
 
 // SendMessage sends message to RabbitMQ if the upload is finished
 func (m *AMQPMessenger) SendMessage(corrID string, body []byte) error {
+	if m.channel.IsClosed() {
+		log.Debugln("channel closed, reconnecting")
+		if err := m.createNewChannel(); err != nil {
+			return fmt.Errorf("failed to recreate channel: %v", err)
+		}
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	err := m.channel.PublishWithContext(
@@ -133,3 +140,20 @@ func buildMqURI(mqHost, mqPort, mqUser, mqPassword, mqVhost string, ssl bool) st
 
 	return brokerURI
 }
+
+func (m *AMQPMessenger) createNewChannel() error {
+	c, err := m.connection.Channel()
+	if err != nil {
+		return err
+	}
+	confirmsChan := make(chan amqp.Confirmation, 1)
+	if err := c.Confirm(false); err != nil {
+		close(confirmsChan)
+		log.Fatalf("Channel could not be put into confirm mode: %s\n", err)
+	}
+	log.Debugln("recconected to new channel")
+	m.channel = c
+	m.confirmsChan = c.NotifyPublish(confirmsChan)
+
+	return nil
+}

messenger_test.go

@@ -67,3 +67,37 @@ func TestSendMessage(t *testing.T) {
 	err = messenger.SendMessage(uuid.String(), jsonMessage)
 	assert.NoError(t, err)
 }
+
+func TestCreateNewChannel(t *testing.T) {
+	viper.Reset()
+	viper.Set("server.confFile", "dev_utils/config.yaml")
+
+	config, err := NewConfig()
+	assert.NotNil(t, config)
+	assert.NoError(t, err)
+	tlsConfig, err := TLSConfigBroker(config)
+	if err != nil {
+		t.Log(err)
+		t.Skip("skip test since certificates are not present")
+	}
+	assert.NotNil(t, tlsConfig)
+	assert.NoError(t, err)
+
+	messenger, err := NewAMQPMessenger(config.Broker, tlsConfig)
+	messenger.channel.Close()
+	assert.NoError(t, err)
+	event := Event{}
+	checksum := Checksum{}
+	event.Operation = "TestRecreateChannel"
+	event.Username = "Dummy"
+	checksum.Type = "md5"
+	checksum.Value = "123456789"
+	event.Checksum = []interface{}{checksum}
+
+	jsonMessage, err := json.Marshal(event)
+	assert.NoError(t, err)
+	uuid, _ := uuid.NewRandom()
+	t.Log("uuid: ", uuid)
+	err = messenger.SendMessage(uuid.String(), jsonMessage)
+	assert.NoError(t, err)
+}