Service architecture refactoring (ortuman#104)

Author: ortuman

.travis.yml

@@ -10,7 +10,7 @@ os:
   - osx
 
 go:
-  - 1.13.x
+  - 1.14.x
 
 before_script:
   - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then curl -L https://codeclimate.com/downloads/test-reporter/test-reporter-latest-darwin-amd64 > ./cc-test-reporter; fi

CHANGELOG.md

@@ -4,11 +4,7 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
-## [0.8.2] - 2020-01-14
-### Fixed
-- Blocking Command implementation
-
-## [0.8.1] - 2020-01-07
-### Added
-- Added context support
+## [0.9.0] - 2020-02-15
+### Changed
+- Router implementation refactoring
 

Makefile

@@ -1,8 +1,8 @@
 install:
-	@export GO111MODULE=on && go install -ldflags="-s -w" github.com/ortuman/jackal
+	@go install -ldflags="-s -w" github.com/ortuman/jackal
 
 install-tools:
-	@export GO111MODULE=on && go get -u \
+	@go get -u \
 		golang.org/x/lint/golint \
 		golang.org/x/tools/cmd/goimports
 
@@ -14,7 +14,7 @@ fmt: install-tools
 
 build:
 	@echo "Building binary..."
-	@export GO111MODULE=on && go build -ldflags="-s -w"
+	@export go build -ldflags="-s -w"
 
 test:
 	@echo "Running tests..."
@@ -34,7 +34,7 @@ lint: install-tools
 
 dockerimage:
 	@echo "Building binary..."
-	@env GO111MODULE=on GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -ldflags="-s -w"
+	@env GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -ldflags="-s -w"
 	@echo "Building docker image..."
 	@docker build -f dockerfiles/Dockerfile -t ortuman/jackal .
 

README.md

@@ -128,22 +128,6 @@ However there's a chance to forward offline messages to some external service by
 
 Each time a message is sent to an offline user a `POST` http request to the `pass` URL is made, using the specified `Authorization` header and including the message stanza into the request body.
 
-## Cluster configuration
-
-The purpose of clustering is to be able to use several servers for fault-tolerance and scalability.
-
-To run `jackal` in clustering mode make sure to add a `cluster` section configuration in each of the cluster nodes.
-
-Here is an example of how this section should look like:
-```yaml
-cluster:
-  name: node1                             
-  port: 5010                              
-  hosts: [node2:5010, node3:5010] 
-```
-
-Do not forget to include all cluster nodes, excluding the local one, in the `hosts` array field. Otherwise the expected behavior will be undefined.   
-
 ## Run jackal in Docker
 
 Set up `jackal` in the cloud in under 5 minutes with zero knowledge of Golang or Linux shell using our [jackal Docker image](https://hub.docker.com/r/ortuman/jackal/).

app/app.go

@@ -20,19 +20,24 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/ortuman/jackal/c2s"
-	"github.com/ortuman/jackal/cluster"
+	c2srouter "github.com/ortuman/jackal/c2s/router"
 	"github.com/ortuman/jackal/component"
 	"github.com/ortuman/jackal/log"
 	"github.com/ortuman/jackal/module"
 	"github.com/ortuman/jackal/router"
+	"github.com/ortuman/jackal/router/host"
 	"github.com/ortuman/jackal/s2s"
+	s2srouter "github.com/ortuman/jackal/s2s/router"
 	"github.com/ortuman/jackal/storage"
 	"github.com/ortuman/jackal/version"
 	"github.com/pkg/errors"
 )
 
 const (
+	envAllocationID = "JACKAL_ALLOCATION_ID"
+
 	defaultShutDownWaitTime = time.Duration(5) * time.Second
 )
 
@@ -60,10 +65,10 @@ type Application struct {
 	output           io.Writer
 	args             []string
 	logger           log.Logger
-	cluster          *cluster.Cluster
-	router           *router.Router
+	router           router.Router
 	mods             *module.Modules
 	comps            *component.Components
+	s2sOutProvider   *s2s.OutProvider
 	s2s              *s2s.S2S
 	c2s              *c2s.C2S
 	debugSrv         *http.Server
@@ -121,6 +126,7 @@ func (a *Application) Run() error {
 	if err != nil {
 		return err
 	}
+
 	// create PID file
 	if err := a.createPIDFile(cfg.PIDFile); err != nil {
 		return err
@@ -131,50 +137,56 @@ func (a *Application) Run() error {
 		return err
 	}
 
+	// set allocation identifier
+	allocID := os.Getenv(envAllocationID)
+	if len(allocID) == 0 {
+		allocID = uuid.New().String()
+	}
+
 	// show jackal's fancy logo
-	a.printLogo()
+	a.printLogo(allocID)
 
 	// initialize storage
 	repContainer, err := storage.New(&cfg.Storage)
 	if err != nil {
 		return err
 	}
-	// initialize router
-	a.router, err = router.New(&cfg.Router, repContainer.User(), repContainer.BlockList())
+	if err := repContainer.Presences().ClearPresences(context.Background()); err != nil {
+		return err
+	}
+
+	// initialize hosts
+	hosts, err := host.New(cfg.Hosts)
 	if err != nil {
 		return err
 	}
+	// initialize router
+	var s2sRouter router.S2SRouter
 
-	// initialize cluster
-	if cfg.Cluster != nil {
-		if repContainer.IsClusterCompatible() {
-			a.cluster, err = cluster.New(cfg.Cluster, a.router.ClusterDelegate())
-			if err != nil {
-				return err
-			}
-			if a.cluster != nil {
-				a.router.SetCluster(a.cluster)
-				if err := a.cluster.Join(); err != nil {
-					log.Warnf("%v", err)
-				}
-			}
-		} else {
-			log.Warnf("cluster mode disabled: storage type '%s' is not compatible", cfg.Storage.Type)
-		}
+	if cfg.S2S != nil {
+		a.s2sOutProvider = s2s.NewOutProvider(cfg.S2S, hosts)
+		s2sRouter = s2srouter.New(a.s2sOutProvider)
+	}
+	a.router, err = router.New(
+		hosts,
+		c2srouter.New(repContainer.User(), repContainer.BlockList()),
+		s2sRouter,
+	)
+	if err != nil {
+		return err
 	}
 
 	// initialize modules & components...
-	a.mods = module.New(&cfg.Modules, a.router, repContainer)
+	a.mods = module.New(&cfg.Modules, a.router, repContainer, allocID)
 	a.comps = component.New(&cfg.Components, a.mods.DiscoInfo)
 
 	// start serving s2s...
-	a.s2s = s2s.New(cfg.S2S, a.mods, a.router)
-	if a.s2s != nil {
-		a.router.SetOutS2SProvider(a.s2s)
+	if cfg.S2S != nil {
+		a.s2s = s2s.New(cfg.S2S, a.mods, a.s2sOutProvider, a.router)
 		a.s2s.Start()
 	}
 	// start serving c2s...
-	a.c2s, err = c2s.New(cfg.C2S, a.mods, a.comps, a.router, repContainer.User())
+	a.c2s, err = c2s.New(cfg.C2S, a.mods, a.comps, a.router, repContainer.User(), repContainer.BlockList())
 	if err != nil {
 		return err
 	}
@@ -240,12 +252,12 @@ func (a *Application) initLogger(config *loggerConfig, output io.Writer) error {
 	return nil
 }
 
-func (a *Application) printLogo() {
+func (a *Application) printLogo(allocID string) {
 	for i := range logoStr {
 		log.Infof("%s", logoStr[i])
 	}
 	log.Infof("")
-	log.Infof("jackal %v\n", version.ApplicationVersion)
+	log.Infof("jackal %v - allocation_id: %s\n", version.ApplicationVersion, allocID)
 }
 
 func (a *Application) initDebugServer(port int) error {
@@ -280,21 +292,34 @@ func (a *Application) gracefullyShutdown() error {
 func (a *Application) shutdown(ctx context.Context) <-chan bool {
 	c := make(chan bool, 1)
 	go func() {
-		if a.debugSrv != nil {
-			_ = a.debugSrv.Shutdown(ctx)
-		}
-		a.c2s.Shutdown(ctx)
-		if a.s2s != nil {
-			a.s2s.Shutdown(ctx)
+		if err := a.doShutdown(ctx); err != nil {
+			log.Warnf("failed to shutdown: %s", err)
 		}
-		if a.cluster != nil {
-			_ = a.cluster.Shutdown(ctx)
-		}
-		_ = a.comps.Shutdown(ctx)
-		_ = a.mods.Shutdown(ctx)
-
-		log.Unset()
 		c <- true
 	}()
 	return c
 }
+
+func (a *Application) doShutdown(ctx context.Context) error {
+	if a.debugSrv != nil {
+		if err := a.debugSrv.Shutdown(ctx); err != nil {
+			return err
+		}
+	}
+	a.c2s.Shutdown(ctx)
+
+	if err := a.comps.Shutdown(ctx); err != nil {
+		return err
+	}
+	if err := a.mods.Shutdown(ctx); err != nil {
+		return err
+	}
+
+	if outProvider := a.s2sOutProvider; outProvider != nil {
+		if err := outProvider.Shutdown(ctx); err != nil {
+			return err
+		}
+	}
+	log.Unset()
+	return nil
+}

app/config.go

@@ -9,15 +9,13 @@ import (
 	"bytes"
 	"io/ioutil"
 
-	"github.com/ortuman/jackal/cluster"
-
 	"github.com/ortuman/jackal/c2s"
 	"github.com/ortuman/jackal/component"
 	"github.com/ortuman/jackal/module"
-	"github.com/ortuman/jackal/router"
+	"github.com/ortuman/jackal/router/host"
 	"github.com/ortuman/jackal/s2s"
 	"github.com/ortuman/jackal/storage"
-	yaml "gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v2"
 )
 
 // debugConfig represents debug server configuration.
@@ -36,16 +34,14 @@ type Config struct {
 	Debug      debugConfig      `yaml:"debug"`
 	Logger     loggerConfig     `yaml:"logger"`
 	Storage    storage.Config   `yaml:"storage"`
-	Cluster    *cluster.Config  `yaml:"cluster"`
-	Router     router.Config    `yaml:"router"`
+	Hosts      []host.Config    `yaml:"hosts"`
 	Modules    module.Config    `yaml:"modules"`
 	Components component.Config `yaml:"components"`
 	C2S        []c2s.Config     `yaml:"c2s"`
 	S2S        *s2s.Config      `yaml:"s2s"`
 }
 
-// FromFile loads default global configuration from
-// a specified file.
+// FromFile loads default global configuration from a specified file.
 func (cfg *Config) FromFile(configFile string) error {
 	b, err := ioutil.ReadFile(configFile)
 	if err != nil {
@@ -54,8 +50,7 @@ func (cfg *Config) FromFile(configFile string) error {
 	return yaml.Unmarshal(b, cfg)
 }
 
-// FromBuffer loads default global configuration from
-// a specified byte buffer.
+// FromBuffer loads default global configuration from a specified byte buffer.
 func (cfg *Config) FromBuffer(buf *bytes.Buffer) error {
 	return yaml.Unmarshal(buf.Bytes(), cfg)
 }

auth/auth_test.go

@@ -9,11 +9,11 @@ import (
 	"context"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/ortuman/jackal/model"
 	memorystorage "github.com/ortuman/jackal/storage/memory"
 	"github.com/ortuman/jackal/stream"
 	"github.com/ortuman/jackal/xmpp/jid"
-	"github.com/pborman/uuid"
 	"github.com/stretchr/testify/require"
 )
 
@@ -24,7 +24,7 @@ func authTestSetup(user *model.User) (*stream.MockC2S, *memorystorage.User) {
 
 	j, _ := jid.New("mariana", "localhost", "res", true)
 
-	testStm := stream.NewMockC2S(uuid.New(), j)
+	testStm := stream.NewMockC2S(uuid.New().String(), j)
 
 	testStm.SetJID(j)
 	return testStm, s

auth/scram.go

@@ -17,14 +17,14 @@ import (
 	"hash"
 	"strings"
 
+	"github.com/google/uuid"
 	"github.com/ortuman/jackal/model"
 	"github.com/ortuman/jackal/storage/repository"
 	"github.com/ortuman/jackal/stream"
 	"github.com/ortuman/jackal/transport"
 	utilrand "github.com/ortuman/jackal/util/rand"
 	utilstring "github.com/ortuman/jackal/util/string"
 	"github.com/ortuman/jackal/xmpp"
-	"github.com/pborman/uuid"
 	"golang.org/x/crypto/pbkdf2"
 )
 
@@ -222,7 +222,7 @@ func (s *Scram) handleStart(ctx context.Context, elem xmpp.XElement) error {
 	}
 	s.user = user
 
-	s.srvNonce = cNonce + "-" + uuid.New()
+	s.srvNonce = cNonce + "-" + uuid.New().String()
 	s.salt = utilrand.RandomBytes(32)
 	sb64 := base64.StdEncoding.EncodeToString(s.salt)
 	s.firstMessage = fmt.Sprintf("r=%s,s=%s,i=%d", s.srvNonce, sb64, iterationsCount)

c2s/c2s.go

@@ -33,9 +33,7 @@ type c2sServer interface {
 	shutdown(ctx context.Context) error
 }
 
-var createC2SServer = func(config *Config, mods *module.Modules, comps *component.Components, router *router.Router, userRep repository.User) c2sServer {
-	return &server{cfg: config, mods: mods, comps: comps, router: router, userRep: userRep}
-}
+var createC2SServer = newC2SServer
 
 // C2S represents a client-to-server connection manager.
 type C2S struct {
@@ -45,13 +43,13 @@ type C2S struct {
 }
 
 // New returns a new instance of a c2s connection manager.
-func New(configs []Config, mods *module.Modules, comps *component.Components, router *router.Router, userRep repository.User) (*C2S, error) {
+func New(configs []Config, mods *module.Modules, comps *component.Components, router router.Router, userRep repository.User, blockListRep repository.BlockList) (*C2S, error) {
 	if len(configs) == 0 {
 		return nil, errors.New("at least one c2s configuration is required")
 	}
 	c := &C2S{servers: make(map[string]c2sServer)}
 	for _, config := range configs {
-		srv := createC2SServer(&config, mods, comps, router, userRep)
+		srv := createC2SServer(&config, mods, comps, router, userRep, blockListRep)
 		c.servers[config.ID] = srv
 	}
 	return c, nil