Merge branches 'fix/not-starting-kafkaproducer-in-integration-tests' …

…and 'master' of https://github.com/cosminrentea/gobbler into fix/not-starting-kafkaproducer-in-integration-tests

# Conflicts:
#	server/configstring/configstringlist_test.go

server/config.go

@@ -226,7 +226,7 @@ var (
 				Default(strconv.Itoa(runtime.NumCPU())).
 				Envar("GUBLE_SMS_WORKERS").
 				Int(),
-			KafkaReportingTopic:kingpin.Flag("sms-kafka-topic", "The name of the SMS-Reporting Kafka topic").
+			KafkaReportingTopic: kingpin.Flag("sms-kafka-topic", "The name of the SMS-Reporting Kafka topic").
 				Envar("GUBLE_SMS_KAFKA_TOPIC").
 				String(),
 			IntervalMetrics: &defaultSMSMetrics,

server/configstring/configstringlist_test.go

@@ -1,10 +1,10 @@
 package configstring
 
 import (
-	"testing"
 	"github.com/cosminrentea/gobbler/testutil"
-	"github.com/stretchr/testify/assert"
 	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	"testing"
 )
 
 func TestList_IsEmpty(t *testing.T) {
@@ -21,4 +21,4 @@ func TestList_IsEmpty(t *testing.T) {
 	list := &List{}
 
 	assert.True(t, list.IsEmpty())
-}
+}

server/connector/mocks_connector_gen_test.go

@@ -6,7 +6,7 @@ package connector
 import (
 	context "context"
 	protocol "github.com/cosminrentea/gobbler/protocol"
-	
+
 	router "github.com/cosminrentea/gobbler/server/router"
 	gomock "github.com/golang/mock/gomock"
 	http "net/http"

server/router/router.go

@@ -154,9 +154,9 @@ func (router *router) Check() error {
 // and then passes it to the internal channel, and asynchronously to the cluster (if available).
 func (router *router) HandleMessage(message *protocol.Message) error {
 	logger.WithFields(log.Fields{
-		"userID":        message.UserID,
+		"userID":         message.UserID,
 		"correlation_id": message.CorrelationID(),
-		"path":          message.Path}).Debug("HandleMessage")
+		"path":           message.Path}).Debug("HandleMessage")
 
 	mTotalMessagesIncoming.Add(1)
 	pMessagesIncoming.Inc()

server/sms/nexmo_sms_sender.go

@@ -29,7 +29,7 @@ const (
 	ResponseNumberBarred
 	ResponsePartnerAcctBarred
 	ResponsePartnerQuotaExceeded
-	ResponseUnused
+	ResponseUnused //Defined in Nexmo Api
 	ResponseRESTNotEnabled
 	ResponseMessageTooLong
 	ResponseCommunicationFailed
@@ -45,14 +45,14 @@ var (
 	MaxIdleConnections = 100
 	RequestTimeout     = 500 * time.Millisecond
 
-	ErrHTTPClientError           = errors.New("Http client sending to Nexmo Failed.No sms was sent")
-	ErrNexmoResponseStatusNotOk  = errors.New("Nexmo response status not ResponseSuccess")
-	ErrSMSResponseDecodingFailed = errors.New("Nexmo response decoding failed")
-	ErrInvalidSender             = errors.New("Sms destination phoneNumber is invalid")
-	ErrMultipleSmsSent           = errors.New("Multiple  or no sms we're sent.SMS message may be too long")
-	ErrRetryFailed               = errors.New("Failed retrying to send message")
-	ErrEncodeFailed              = errors.New("Encoding of message to be sent to Nexmo  failed")
-
+	ErrHTTPClientError             = errors.New("Http client sending to Nexmo Failed.No sms was sent")
+	ErrNexmoResponseStatusNotOk    = errors.New("Nexmo response status not ResponseSuccess")
+	ErrSMSResponseDecodingFailed   = errors.New("Nexmo response decoding failed")
+	ErrInvalidSender               = errors.New("Sms destination phoneNumber is invalid")
+	ErrMultipleSmsSent             = errors.New("Multiple  or no sms we're sent.SMS message may be too long")
+	ErrRetryFailed                 = errors.New("Failed retrying to send message")
+	ErrEncodeFailed                = errors.New("Encoding of message to be sent to Nexmo  failed")
+	ErrLastIDCouldNotBeSet         = errors.New("Setting last id failed")
 	errKafkaReportingConfiguration = errors.New("Kafka Reporting for Nexmo is not correctly configured")
 )
 

server/sms/sms_gateway.go

@@ -166,7 +166,7 @@ func (g *gateway) proxyLoop() error {
 			}
 
 			err := g.send(receivedMsg)
-			if err != nil && err == ErrRetryFailed {
+			if err == ErrRetryFailed || err == ErrLastIDCouldNotBeSet {
 				// THIS MAY BE BLOCKING.Maybe not a good idea.
 				for err2 := g.SetLastSentID(receivedMsg.ID); err2 != nil; {
 					g.logger.WithField("error", err2.Error()).Error("Error setting last ID.Retrying")
@@ -196,7 +196,12 @@ func (g *gateway) send(receivedMsg *protocol.Message) error {
 	}
 	mTotalSentMessages.Add(1)
 	pSent.Inc()
-	g.SetLastSentID(receivedMsg.ID)
+	err = g.SetLastSentID(receivedMsg.ID)
+	if err != nil {
+		logger.WithField("error", err.Error()).Error("Setting last id for sms connector failed.")
+		return ErrLastIDCouldNotBeSet
+	}
+
 	return nil
 }