Allow up to max_inflight many concurrent function invocations.

Fixes #94

Switch from auto ack to manual ack.
This enables message handling in individual go routines.
In main.go switch to atomic counter and include index in traces.
Going forward these may be mixed with concurrent traces.

Removes option to run without durable.
Generates durable name base on subject.
Removes issue where the durable would not see every messages.

Signed-off-by: Matthias Hanel <mh@synadia.com>

Author: matthiashanel

main.go

@@ -14,6 +14,7 @@ import (
 	"os/signal"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	stan "github.com/nats-io/stan.go"
@@ -34,12 +35,6 @@ func main() {
 
 	hostname, _ := os.Hostname()
 
-	var durable string
-	if config.NatsDurableQueueSubscription {
-		durable = "faas"
-	}
-
-	var unsubscribe bool
 	var credentials *auth.BasicAuthCredentials
 	var err error
 
@@ -53,9 +48,9 @@ func main() {
 
 	client := makeClient(config.TLSInsecure)
 
-	i := 0
+	counter := uint64(0)
 	messageHandler := func(msg *stan.Msg) {
-		i++
+		i := atomic.AddUint64(&counter, 1)
 
 		log.Printf("[#%d] Received on [%s]: '%s'\n", i, msg.Subject, msg)
 
@@ -65,14 +60,14 @@ func main() {
 		unmarshalErr := json.Unmarshal(msg.Data, &req)
 
 		if unmarshalErr != nil {
-			log.Printf("Unmarshal error: %s with data %s", unmarshalErr, msg.Data)
+			log.Printf("[#%d] Unmarshal error: %s with data %s", i, unmarshalErr, msg.Data)
 			return
 		}
 
 		xCallID := req.Header.Get("X-Call-Id")
 
 		functionURL := makeFunctionURL(&req, &config, req.Path, req.QueryString)
-		fmt.Printf("Invoking: %s with %d bytes, via: %s\n", req.Function, len(req.Body), functionURL)
+		fmt.Printf("[#%d] Invoking: %s with %d bytes, via: %s\n", i, req.Function, len(req.Body), functionURL)
 
 		if config.DebugPrintBody {
 			fmt.Println(string(req.Body))
@@ -81,7 +76,7 @@ func main() {
 		start := time.Now()
 		request, err := http.NewRequest(http.MethodPost, functionURL, bytes.NewReader(req.Body))
 		if err != nil {
-			log.Printf("Unable to post message due to invalid URL, error: %s", err.Error())
+			log.Printf("[#%d] Unable to post message due to invalid URL, error: %s", i, err.Error())
 			return
 		}
 
@@ -102,12 +97,12 @@ func main() {
 
 		duration := time.Since(start)
 
-		log.Printf("Invoked: %s [%d] in %fs", req.Function, statusCode, duration.Seconds())
+		log.Printf("[#%d] Invoked: %s [%d] in %fs", i, req.Function, statusCode, duration.Seconds())
 
 		if err != nil {
 			status = http.StatusServiceUnavailable
 
-			log.Printf("Error invoking %s, error: %s", req.Function, err)
+			log.Printf("[#%d] Error invoking %s, error: %s", i, req.Function, err)
 
 			timeTaken := time.Since(started).Seconds()
 
@@ -122,18 +117,18 @@ func main() {
 					timeTaken)
 
 				if resultErr != nil {
-					log.Printf("Posted callback to: %s - status %d, error: %s\n", req.CallbackURL.String(), http.StatusServiceUnavailable, resultErr.Error())
+					log.Printf("[#%d] Posted callback to: %s - status %d, error: %s\n", i, req.CallbackURL.String(), http.StatusServiceUnavailable, resultErr.Error())
 				} else {
-					log.Printf("Posted result to %s - status: %d", req.CallbackURL.String(), resultStatusCode)
+					log.Printf("[#%d] Posted result to %s - status: %d", i, req.CallbackURL.String(), resultStatusCode)
 				}
 			}
 
 			if config.GatewayInvoke == false {
 				statusCode, reportErr := postReport(&client, req.Function, status, timeTaken, config.GatewayAddressURL(), credentials)
 				if reportErr != nil {
-					log.Printf("Error posting report: %s\n", reportErr)
+					log.Printf("[#%d] Error posting report: %s\n", i, reportErr)
 				} else {
-					log.Printf("Posting report to gateway for %s - status: %d\n", req.Function, statusCode)
+					log.Printf("[#%d] Posting report to gateway for %s - status: %d\n", i, req.Function, statusCode)
 				}
 				return
 			}
@@ -148,20 +143,20 @@ func main() {
 			functionResult = resData
 
 			if err != nil {
-				log.Printf("Error reading body for: %s, error: %s", req.Function, err)
+				log.Printf("[#%d] Error reading body for: %s, error: %s",i,  req.Function, err)
 			}
 
 			if config.WriteDebug {
 				fmt.Println(string(functionResult))
 			} else {
-				fmt.Printf("%s returned %d bytes\n", req.Function, len(functionResult))
+				fmt.Printf("[#%d] %s returned %d bytes\n", i, req.Function, len(functionResult))
 			}
 		}
 
 		timeTaken := time.Since(started).Seconds()
 
 		if req.CallbackURL != nil {
-			log.Printf("Callback to: %s\n", req.CallbackURL.String())
+			log.Printf("[#%d] Callback to: %s\n", i, req.CallbackURL.String())
 
 			resultStatusCode, resultErr := postResult(&client,
 				res,
@@ -173,18 +168,18 @@ func main() {
 				timeTaken)
 
 			if resultErr != nil {
-				log.Printf("Error posting to callback-url: %s\n", resultErr)
+				log.Printf("[#%d] Error posting to callback-url: %s\n", i, resultErr)
 			} else {
-				log.Printf("Posted result for %s to callback-url: %s, status: %d", req.Function, req.CallbackURL.String(), resultStatusCode)
+				log.Printf("[#%d] Posted result for %s to callback-url: %s, status: %d", i, req.Function, req.CallbackURL.String(), resultStatusCode)
 			}
 		}
 
 		if config.GatewayInvoke == false {
 			statusCode, reportErr := postReport(&client, req.Function, res.StatusCode, timeTaken, config.GatewayAddressURL(), credentials)
 			if reportErr != nil {
-				log.Printf("Error posting report: %s\n", reportErr.Error())
+				log.Printf("[#%d] Error posting report: %s\n", i, reportErr.Error())
 			} else {
-				log.Printf("Posting report for %s, status: %d\n", req.Function, statusCode)
+				log.Printf("[#%d] Posting report for %s, status: %d\n", i, req.Function, statusCode)
 			}
 		}
 
@@ -204,10 +199,8 @@ func main() {
 
 		subject:        config.NatsChannel,
 		qgroup:         config.NatsQueueGroup,
-		durable:        durable,
 		messageHandler: messageHandler,
-		startOption:    stan.StartWithLastReceived(),
-		maxInFlight:    stan.MaxInflight(config.MaxInflight),
+		maxInFlight:    config.MaxInflight,
 		ackWait:        config.AckWait,
 	}
 
@@ -218,29 +211,13 @@ func main() {
 	// Wait for a SIGINT (perhaps triggered by user with CTRL-C)
 	// Run cleanup when signal is received
 	signalChan := make(chan os.Signal, 1)
-	cleanupDone := make(chan bool)
 	signal.Notify(signalChan, os.Interrupt)
-	go func() {
-		for range signalChan {
-			fmt.Printf("\nReceived an interrupt, unsubscribing and closing connection...\n\n")
-			// Do not unsubscribe a durable on exit, except if asked to.
-			if durable == "" || unsubscribe {
-				if err := natsQueue.unsubscribe(); err != nil {
-					log.Panicf(
-						"Cannot unsubscribe subject: %s from %s because of an error: %v",
-						natsQueue.subject,
-						natsQueue.natsURL,
-						err,
-					)
-				}
-			}
-			if err := natsQueue.closeConnection(); err != nil {
-				log.Panicf("Cannot close connection to %s because of an error: %v\n", natsQueue.natsURL, err)
-			}
-			cleanupDone <- true
-		}
-	}()
-	<-cleanupDone
+	<-signalChan
+	fmt.Printf("\nReceived an interrupt, unsubscribing and closing connection...\n\n")
+	if err := natsQueue.closeConnection(); err != nil {
+		log.Panicf("Cannot close connection to %s because of an error: %v\n", natsQueue.natsURL, err)
+	}
+	close(signalChan)
 }
 
 // makeClient constructs a HTTP client with keep-alive turned

readconfig.go

@@ -45,14 +45,6 @@ func (ReadConfig) Read() (QueueWorkerConfig, error) {
 		cfg.NatsClusterName = "faas-cluster"
 	}
 
-	if val, exists := os.LookupEnv("faas_nats_durable_queue_subscription"); exists {
-		if v, err := strconv.ParseBool(val); err == nil {
-			cfg.NatsDurableQueueSubscription = v
-		} else {
-			cfg.NatsDurableQueueSubscription = false
-		}
-	}
-
 	if val, exists := os.LookupEnv("faas_nats_channel"); exists && val != "" {
 		cfg.NatsChannel = val
 	} else {
@@ -173,7 +165,6 @@ type QueueWorkerConfig struct {
 	NatsAddress                  string
 	NatsPort                     int
 	NatsClusterName              string
-	NatsDurableQueueSubscription bool
 	NatsChannel                  string
 	NatsQueueGroup               string
 

readconfig_test.go

@@ -101,7 +101,6 @@ func Test_ReadConfig(t *testing.T) {
 	os.Setenv("faas_nats_address", "test_nats")
 	os.Setenv("faas_nats_port", "1234")
 	os.Setenv("faas_nats_cluster_name", "example-nats-cluster")
-	os.Setenv("faas_nats_durable_queue_subscription", "true")
 	os.Setenv("faas_nats_channel", "foo")
 	os.Setenv("faas_nats_queue_group", "bar")
 	os.Setenv("faas_gateway_address", "test_gatewayaddr")
@@ -132,11 +131,6 @@ func Test_ReadConfig(t *testing.T) {
 		t.Fail()
 	}
 
-	wantNatsDurableQueueSubscription := true
-	if config.NatsDurableQueueSubscription != wantNatsDurableQueueSubscription {
-		t.Logf("NatsDurableQueueSubscription want `%t`, got `%t`\n", wantNatsDurableQueueSubscription, config.NatsDurableQueueSubscription)
-	}
-
 	want = "foo"
 	if config.NatsChannel != want {
 		t.Logf("NatsChannel want `%s`, got `%s`\n", want, config.NatsChannel)

types.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 	"log"
+	"strings"
 	"sync"
 	"time"
 
@@ -30,12 +31,11 @@ type NATSQueue struct {
 
 	subject        string
 	qgroup         string
-	durable        string
 	ackWait        time.Duration
 	messageHandler func(*stan.Msg)
-	startOption    stan.SubscriptionOption
-	maxInFlight    stan.SubscriptionOption
+	maxInFlight    int
 	subscription   stan.Subscription
+	msgChan        chan *stan.Msg
 }
 
 // connect creates a subscription to NATS Streaming
@@ -64,29 +64,52 @@ func (q *NATSQueue) connect() error {
 	log.Printf("Subscribing to: %s at %s\n", q.subject, q.natsURL)
 	log.Println("Wait for ", q.ackWait)
 
+	// Pre-fill chan with q.maxInFlight tokens
+	msgChan := make(chan *stan.Msg)
+
+	handler := q.messageHandler
+	opts := []stan.SubscriptionOption{
+		stan.DurableName(strings.ReplaceAll(q.subject, ".", "_")),
+		stan.AckWait(q.ackWait),
+		stan.DeliverAllAvailable(),
+		stan.MaxInflight(q.maxInFlight),
+	}
+	if q.maxInFlight > 1 {
+		for i := 0; i < q.maxInFlight; i++ {
+			go func() {
+				for msg := range msgChan {
+					q.messageHandler(msg)
+					msg.Ack()
+				}
+			}()
+		}
+
+		opts = append(opts, stan.SetManualAckMode())
+		handler = func(msg *stan.Msg) {
+			msgChan <- msg
+		}
+	}
 	subscription, err := q.conn.QueueSubscribe(
 		q.subject,
 		q.qgroup,
-		q.messageHandler,
-		stan.DurableName(q.durable),
-		stan.AckWait(q.ackWait),
-		q.startOption,
-		q.maxInFlight,
+		handler,
+		opts...
 	)
 
 	if err != nil {
 		return fmt.Errorf("couldn't subscribe to %s at %s. Error: %v", q.subject, q.natsURL, err)
 	}
 
 	log.Printf(
-		"Listening on [%s], clientID=[%s], qgroup=[%s] durable=[%s]\n",
+		"Listening on [%s], clientID=[%s], qgroup=[%s] maxInFlight=[%d]\n",
 		q.subject,
 		q.clientID,
 		q.qgroup,
-		q.durable,
+		q.maxInFlight,
 	)
 
 	q.subscription = subscription
+	q.msgChan = msgChan
 
 	return nil
 }
@@ -117,17 +140,6 @@ func (q *NATSQueue) reconnect() {
 	log.Printf("Reconnecting limit (%d) reached\n", q.maxReconnect)
 }
 
-func (q *NATSQueue) unsubscribe() error {
-	q.connMutex.Lock()
-	defer q.connMutex.Unlock()
-
-	if q.subscription != nil {
-		return fmt.Errorf("q.subscription is nil")
-	}
-
-	return q.subscription.Unsubscribe()
-}
-
 func (q *NATSQueue) closeConnection() error {
 	q.connMutex.Lock()
 	defer q.connMutex.Unlock()
@@ -136,7 +148,9 @@ func (q *NATSQueue) closeConnection() error {
 		return fmt.Errorf("q.conn is nil")
 	}
 
+	err := q.conn.Close()
+	close(q.msgChan)
 	close(q.quitCh)
 
-	return q.conn.Close()
+	return err
 }