Implement storageservice in nats subscription object
This commit is contained in:
parent
7c7eff2aff
commit
f441047fa1
25
cmd/main.go
25
cmd/main.go
@ -2,13 +2,14 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"git.defalsify.org/vise.git/db/mem"
|
||||
"git.grassecon.net/urdt/ussd/initializers"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
"git.grassecon.net/term/config"
|
||||
"git.grassecon.net/term/event/nats"
|
||||
)
|
||||
@ -18,14 +19,28 @@ func init() {
|
||||
}
|
||||
|
||||
func main() {
|
||||
config.LoadConfig()
|
||||
|
||||
var dbDir string
|
||||
flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
|
||||
flag.Parse()
|
||||
|
||||
ctx := context.Background()
|
||||
db := mem.NewMemDb()
|
||||
err := db.Connect(ctx, "")
|
||||
// db := mem.NewMemDb()
|
||||
// err := db.Connect(ctx, "")
|
||||
// if err != nil {
|
||||
// fmt.Fprintf(os.Stderr, "Db connect err: %v", err)
|
||||
// os.Exit(1)
|
||||
// }
|
||||
|
||||
menuStorageService := common.NewStorageService(dbDir)
|
||||
err := menuStorageService.EnsureDbDir()
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Db connect err: %v", err)
|
||||
fmt.Fprintf(os.Stderr, err.Error())
|
||||
os.Exit(1)
|
||||
}
|
||||
n := nats.NewNatsSubscription(db)
|
||||
|
||||
n := nats.NewNatsSubscription(menuStorageService)
|
||||
err = n.Connect(ctx, config.JetstreamURL)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Stream connect err: %v", err)
|
||||
|
@ -9,7 +9,6 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
geEvent "github.com/grassrootseconomics/eth-tracker/pkg/event"
|
||||
"git.defalsify.org/vise.git/logging"
|
||||
"git.defalsify.org/vise.git/db"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
"git.grassecon.net/term/event"
|
||||
"git.grassecon.net/term/config"
|
||||
@ -32,12 +31,10 @@ type NatsSubscription struct {
|
||||
}
|
||||
|
||||
// NewNatsSubscription creates a new NatsSubscription with the given user store.
|
||||
func NewNatsSubscription(store db.Db) *NatsSubscription {
|
||||
func NewNatsSubscription(store common.StorageServices) *NatsSubscription {
|
||||
return &NatsSubscription{
|
||||
Router: event.Router{
|
||||
Store: &common.UserDataStore{
|
||||
Db: store,
|
||||
},
|
||||
Store: store,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
@ -141,8 +141,10 @@ func TestHandleMsg(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
||||
sub := NewNatsSubscription(userDb)
|
||||
storageService := &testutil.TestStorageService{
|
||||
Store: userDb,
|
||||
}
|
||||
sub := NewNatsSubscription(storageService)
|
||||
|
||||
data := fmt.Sprintf(`{
|
||||
"block": %d,
|
||||
|
@ -16,8 +16,7 @@ var (
|
||||
|
||||
// Router is responsible for invoking handlers corresponding to events.
|
||||
type Router struct {
|
||||
// User data store abstraction over application data.
|
||||
Store *common.UserDataStore
|
||||
Store common.StorageServices
|
||||
}
|
||||
|
||||
// Route parses an event from the event stream, and resolves the handler
|
||||
@ -27,13 +26,20 @@ type Router struct {
|
||||
// handler fails to successfully execute.
|
||||
func(r *Router) Route(ctx context.Context, gev *geEvent.Event) error {
|
||||
logg.DebugCtxf(ctx, "have event", "ev", gev)
|
||||
store, err := r.Store.GetUserdataDb(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
userStore := &common.UserDataStore{
|
||||
Db: store,
|
||||
}
|
||||
evCC, ok := asCustodialRegistrationEvent(gev)
|
||||
if ok {
|
||||
return handleCustodialRegistration(ctx, r.Store, evCC)
|
||||
return handleCustodialRegistration(ctx, userStore, evCC)
|
||||
}
|
||||
evTT, ok := asTokenTransferEvent(gev)
|
||||
if ok {
|
||||
return handleTokenTransfer(ctx, r.Store, evTT)
|
||||
return handleTokenTransfer(ctx, userStore, evTT)
|
||||
}
|
||||
|
||||
return fmt.Errorf("unexpected message")
|
||||
|
30
internal/testutil/service.go
Normal file
30
internal/testutil/service.go
Normal file
@ -0,0 +1,30 @@
|
||||
package testutil
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"git.defalsify.org/vise.git/persist"
|
||||
"git.defalsify.org/vise.git/resource"
|
||||
"git.defalsify.org/vise.git/db"
|
||||
)
|
||||
|
||||
type TestStorageService struct {
|
||||
Store db.Db
|
||||
}
|
||||
|
||||
func(ss *TestStorageService) GetUserdataDb(ctx context.Context) (db.Db, error) {
|
||||
return ss.Store, nil
|
||||
}
|
||||
|
||||
func(ts *TestStorageService) GetPersister(ctx context.Context) (*persist.Persister, error) {
|
||||
return nil, errors.New("not implemented")
|
||||
}
|
||||
|
||||
func(ts *TestStorageService) GetResource(ctx context.Context) (resource.Resource, error) {
|
||||
return nil, errors.New("not implemented")
|
||||
}
|
||||
|
||||
func(ss *TestStorageService) EnsureDbDir() error {
|
||||
return nil
|
||||
}
|
Loading…
Reference in New Issue
Block a user