Add linter and fix issues it identified.

Author: SpeedyCoder

.circleci/config.yml

@@ -7,9 +7,13 @@ jobs:
     - image: circleci/golang:1.12
       environment:
         GO111MODULE: "on"
+        GOLANGCI_VERSION: "v1.15.0"
     steps:
+    - run: curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | bash -s ${GOLANGCI_VERSION}
+    - run: mv ./bin/golangci-lint $GOPATH/bin/ && rm -rf bin
     - checkout
     - run: go get -t -d -v ./...
+    - run: golangci-lint run --enable-all -D=lll,gochecknoglobals,gosec
     - run: go test -v -race -cover ./...
     - run: go build -o ./bin/sever ./cmd/proximo-server
     - run: go build -o ./bin/client ./cmd/proximo-client

backend/mem/mem.go

@@ -60,23 +60,27 @@ func (h memBackend) loop() {
 
 		case inm := <-h.incomingMessages:
 			consumers := subs[inm.topic]
-			for _, consumer := range consumers {
-				var remaining []*sub
+
+			remainingConsumers := make(map[string][]*sub, len(consumers))
+			for id, consumer := range consumers {
+				var remainingSubs []*sub
 				sentOne := false
 				for _, sub := range consumer {
 					if !sentOne {
 						select {
 						case <-sub.ctx.Done():
 							// drop expired consumers
 						case sub.msgs <- inm.message:
-							remaining = append(remaining, sub)
+							remainingSubs = append(remainingSubs, sub)
 							sentOne = true
 						}
 					} else {
-						remaining = append(remaining, sub)
+						remainingSubs = append(remainingSubs, sub)
 					}
 				}
+				remainingConsumers[id] = remainingSubs
 			}
+			subs[inm.topic] = remainingConsumers
 
 			h.last100[inm.topic] = append(h.last100[inm.topic], inm.message)
 			for len(h.last100[inm.topic]) > 100 {

cmd/proximo-client/main.go

@@ -10,6 +10,7 @@ import (
 	"syscall"
 
 	"github.com/urfave/cli"
+	"github.com/uw-labs/proximo/internal/id"
 	"github.com/uw-labs/substrate"
 	"github.com/uw-labs/substrate/proximo"
 )
@@ -52,6 +53,9 @@ func main() {
 }
 
 func consume(endpoint string, topic string, consumerID string) error {
+	if consumerID == "" {
+		consumerID = id.Generate()
+	}
 	source, err := proximo.NewAsyncMessageSource(proximo.AsyncMessageSourceConfig{
 		Broker:        endpoint,
 		Topic:         topic,
@@ -106,7 +110,7 @@ func produce(endpoint string, topic string) error {
 func newContext() context.Context {
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
-		sCh := make(chan os.Signal)
+		sCh := make(chan os.Signal, 1)
 		signal.Notify(sCh, os.Interrupt, syscall.SIGTERM)
 		<-sCh
 		cancel()

cmd/proximo-server/main.go

@@ -168,7 +168,10 @@ func main() {
 	if err := app.Run(os.Args); err != nil {
 		log.Fatal(err)
 	}
-	log.Fatal(listenAndServe(sourceFactory, sinkFactory, *port))
+	if err := listenAndServe(sourceFactory, sinkFactory, *port); err != nil {
+		log.Fatal(err)
+	}
+	log.Println("Server terminated cleanly")
 }
 
 func parseEndpoints(endpoints string) map[string]bool {
@@ -186,6 +189,7 @@ func parseEndpoints(endpoints string) map[string]bool {
 
 	return enabled
 }
+
 func listenAndServe(sourceFactory proximo.AsyncSourceFactory, sinkFactory proximo.AsyncSinkFactory, port int) error {
 	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
 	if err != nil {

go.mod

@@ -31,7 +31,7 @@ require (
 	github.com/sirupsen/logrus v1.4.0 // indirect
 	github.com/stretchr/testify v1.2.2
 	github.com/urfave/cli v1.20.0
-	github.com/uw-labs/substrate v0.0.0-20190329123054-36acad44151f
+	github.com/uw-labs/substrate v0.0.0-20190329132135-be046b27cb7d
 	github.com/uw-labs/sync v0.0.0-20190307114256-1bb306bf6e71
 	go.etcd.io/bbolt v1.3.2 // indirect
 	golang.org/x/net v0.0.0-20190318221613-d196dffd7c2b

go.sum

@@ -95,8 +95,8 @@ github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/urfave/cli v1.20.0 h1:fDqGv3UG/4jbVl/QkFwEdddtEDjh/5Ov6X+0B/3bPaw=
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
-github.com/uw-labs/substrate v0.0.0-20190329123054-36acad44151f h1:dZjrXRkei35wu5oHsJjZGJTnLLtjoYRnzEkgiv6XbL0=
-github.com/uw-labs/substrate v0.0.0-20190329123054-36acad44151f/go.mod h1:dyyAUJcMCwdJ5EaYf8bh2rCNvSH9wwQEIpfZwt+bOn8=
+github.com/uw-labs/substrate v0.0.0-20190329132135-be046b27cb7d h1:SN02TsiAK4YQOc/84xJC35RB4n3mi2SEqn2XWh5k0cE=
+github.com/uw-labs/substrate v0.0.0-20190329132135-be046b27cb7d/go.mod h1:dyyAUJcMCwdJ5EaYf8bh2rCNvSH9wwQEIpfZwt+bOn8=
 github.com/uw-labs/sync v0.0.0-20190307114256-1bb306bf6e71 h1:1fL6kfRp9ebXBzK3powp5uKM3dCtK62RCiadQDqcR1Q=
 github.com/uw-labs/sync v0.0.0-20190307114256-1bb306bf6e71/go.mod h1:zTU/pesbF9d3ZN7+XwKhdE50y0SCCblmzwukXRnZkXs=
 go.etcd.io/bbolt v1.3.2 h1:Z/90sZLPOeCy2PwprqkFa25PdkusRzaj9P8zm/KNyvk=

id.go renamed to internal/id/id.go

@@ -1,11 +1,12 @@
-package proximo
+package id
 
 import (
 	"crypto/rand"
 	"encoding/hex"
 )
 
-func generateID() string {
+// Generate generates random string id and panics in case it fails.
+func Generate() string {
 	random := []byte{0, 0, 0, 0, 0, 0, 0, 0}
 	_, err := rand.Read(random)
 	if err != nil {

server_source.go

@@ -8,6 +8,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/uw-labs/proximo/internal/id"
 	"github.com/uw-labs/proximo/proto"
 	"github.com/uw-labs/substrate"
 	"github.com/uw-labs/sync/gogroup"
@@ -130,7 +131,7 @@ func (s *SourceServer) sendMessages(ctx context.Context, stream sendSourceStream
 			return nil
 		case msg := <-messages:
 			aMsg := &ackMsg{
-				id:  generateID(),
+				id:  id.Generate(),
 				msg: msg,
 			}
 			pMsg := &proto.Message{

server_test.go

@@ -3,6 +3,7 @@ package proximo
 import (
 	"context"
 	"fmt"
+	"log"
 	"net"
 	"os"
 	"testing"
@@ -42,7 +43,11 @@ func Setup() error {
 
 	proto.RegisterMessageSourceServer(grpcServer, &SourceServer{SourceFactory: backend})
 	proto.RegisterMessageSinkServer(grpcServer, &SinkServer{SinkFactory: backend})
-	go func() { grpcServer.Serve(lis) }()
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			log.Printf("gRPC server stopped with error: %s", err)
+		}
+	}()
 
 	// Wait for server to start
 	cc, err := grpc.Dial("localhost:6868", grpc.WithInsecure(), grpc.WithBlock())