Refactor archiver

cmd/tdsh-archiver/tdsh-archiver.go

@@ -2,11 +2,12 @@ package main
 
 import (
 	"github.com/creekorful/trandoshan/internal/archiver"
+	"github.com/creekorful/trandoshan/internal/process"
 	"os"
 )
 
 func main() {
-	app := archiver.GetApp()
+	app := process.MakeApp(&archiver.State{})
 	if err := app.Run(os.Args); err != nil {
 		os.Exit(1)
 	}

internal/archiver/archiver.go

@@ -4,82 +4,40 @@ import (
 	"fmt"
 	"github.com/creekorful/trandoshan/internal/archiver/storage"
 	"github.com/creekorful/trandoshan/internal/event"
-	"github.com/creekorful/trandoshan/internal/logging"
-	"github.com/creekorful/trandoshan/internal/util"
+	"github.com/creekorful/trandoshan/internal/process"
 	"github.com/rs/zerolog/log"
-	"github.com/urfave/cli/v2"
-	"os"
-	"os/signal"
 	"strings"
-	"syscall"
 )
 
-// GetApp return the crawler app
-func GetApp() *cli.App {
-	return &cli.App{
-		Name:    "tdsh-archiver",
-		Version: "0.7.0",
-		Usage:   "Trandoshan archiver component",
-		Flags: []cli.Flag{
-			logging.GetLogFlag(),
-			util.GetHubURI(),
-			&cli.StringFlag{
-				Name:     "storage-dir",
-				Usage:    "Path to the storage directory",
-				Required: true,
-			},
-		},
-		Action: execute,
-	}
+type State struct {
+	storage storage.Storage
 }
 
-func execute(ctx *cli.Context) error {
-	logging.ConfigureLogger(ctx)
-
-	log.Info().
-		Str("ver", ctx.App.Version).
-		Str("hub-uri", ctx.String("hub-uri")).
-		Str("storage-dir", ctx.String("storage-dir")).
-		Msg("Starting tdsh-archiver")
+func (state *State) Name() string {
+	return "archiver"
+}
 
-	// Create the subscriber
-	sub, err := event.NewSubscriber(ctx.String("hub-uri"))
-	if err != nil {
-		return err
-	}
-	defer sub.Close()
+func (state *State) FlagsNames() []string {
+	return []string{process.HubURIFlag, process.StorageDirFlag}
+}
 
-	// Create local storage
-	st, err := storage.NewLocalStorage(ctx.String("storage-dir"))
+func (state *State) Provide(provider process.Provider) error {
+	st, err := provider.ArchiverStorage()
 	if err != nil {
 		return err
 	}
-
-	state := state{
-		storage: st,
-	}
-
-	if err := sub.Subscribe(event.NewResourceExchange, "archivingQueue", state.handleNewResourceEvent); err != nil {
-		return err
-	}
-
-	log.Info().Msg("Successfully initialized tdsh-archiver. Waiting for resources")
-
-	// Handle graceful shutdown
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
-
-	// Block until we receive our signal.
-	<-c
+	state.storage = st
 
 	return nil
 }
 
-type state struct {
-	storage storage.Storage
+func (state *State) Subscribers() []process.SubscriberDef {
+	return []process.SubscriberDef{
+		{Exchange: event.NewResourceExchange, Queue: "archivingQueue", Handler: state.handleNewResourceEvent},
+	}
 }
 
-func (state *state) handleNewResourceEvent(subscriber event.Subscriber, msg event.RawMessage) error {
+func (state *State) handleNewResourceEvent(subscriber event.Subscriber, msg event.RawMessage) error {
 	var evt event.NewResourceEvent
 	if err := subscriber.Read(&msg, &evt); err != nil {
 		return err

internal/archiver/archiver_test.go

@@ -30,7 +30,7 @@ func TestHandleNewResourceEvent(t *testing.T) {
 
 	storageMock.EXPECT().Store("https://example.onion", tn, []byte("Server: Traefik\r\nContent-Type: application/html\r\n\r\nHello, world")).Return(nil)
 
-	s := state{storage: storageMock}
+	s := State{storage: storageMock}
 	if err := s.handleNewResourceEvent(subscriberMock, msg); err != nil {
 		t.Fail()
 	}

internal/crawler/crawler.go

@@ -24,7 +24,7 @@ func (state *State) Name() string {
 	return "crawler"
 }
 
-func (state *State) Flags() []string {
+func (state *State) FlagsNames() []string {
 	return []string{process.HubURIFlag, process.TorURIFlag, process.UserAgentFlag, process.ConfigAPIURIFlag}
 }
 

internal/process/process.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"github.com/creekorful/trandoshan/api"
+	"github.com/creekorful/trandoshan/internal/archiver/storage"
 	"github.com/creekorful/trandoshan/internal/clock"
 	configapi "github.com/creekorful/trandoshan/internal/configapi/client"
 	"github.com/creekorful/trandoshan/internal/crawler/http"
@@ -27,6 +28,7 @@ const (
 	TorURIFlag       = "tor-uri"
 	UserAgentFlag    = "user-agent"
 	ConfigAPIURIFlag = "config-api-uri"
+	StorageDirFlag   = "storage-dir"
 )
 
 type Provider interface {
@@ -35,6 +37,7 @@ type Provider interface {
 	APIClient() (api.API, error)
 	FastHTTPClient() (http.Client, error)
 	Subscriber() (event.Subscriber, error)
+	ArchiverStorage() (storage.Storage, error)
 }
 
 type defaultProvider struct {
@@ -78,6 +81,10 @@ func (p *defaultProvider) Subscriber() (event.Subscriber, error) {
 	return event.NewSubscriber(p.ctx.String(HubURIFlag))
 }
 
+func (p *defaultProvider) ArchiverStorage() (storage.Storage, error) {
+	return storage.NewLocalStorage(p.ctx.String(StorageDirFlag))
+}
+
 type SubscriberDef struct {
 	Exchange string
 	Queue    string
@@ -86,7 +93,7 @@ type SubscriberDef struct {
 
 type Process interface {
 	Name() string
-	Flags() []string
+	FlagsNames() []string
 	Provide(provider Provider) error
 	Subscribers() []SubscriberDef
 }
@@ -104,7 +111,7 @@ func MakeApp(process Process) *cli.App {
 
 	// Add custom flags
 	flags := getCustomFlags()
-	for _, flag := range process.Flags() {
+	for _, flag := range process.FlagsNames() {
 		if customFlag, contains := flags[flag]; contains {
 			app.Flags = append(app.Flags, customFlag)
 		}
@@ -187,6 +194,11 @@ func getCustomFlags() map[string]cli.Flag {
 		Usage: "User agent to use",
 		Value: "Mozilla/5.0 (Windows NT 10.0; rv:68.0) Gecko/20100101 Firefox/68.0",
 	}
+	flags[StorageDirFlag] = &cli.StringFlag{
+		Name:     StorageDirFlag,
+		Usage:    "Path to the storage directory",
+		Required: true,
+	}
 
 	return flags
 }

internal/scheduler/scheduler.go

@@ -30,7 +30,7 @@ func (state *State) Name() string {
 	return "scheduler"
 }
 
-func (state *State) Flags() []string {
+func (state *State) FlagsNames() []string {
 	return []string{process.HubURIFlag, process.APIURIFlag, process.APITokenFlag, process.ConfigAPIURIFlag}
 }