Add poll/vote_count as calculated field(#363)

README.md

@@ -163,6 +163,9 @@ The Service uses the following environment variables:
 * `MESSAGE_BUS_PORT`: Port of the redis server. The default is `6379`.
 * `REDIS_TEST_CONN`: Test the redis connection on startup. Disable on the cloud
   if redis needs more time to start then this service. The default is `true`.
+* `VOTE_HOST`: Host of the vote-service. The default is `localhost`.
+* `VOTE_PORT`: Port of the vote-service. The default is `9013`.
+* `VOTE_PROTOCAL`: Protocol of the vote-service. The default is `http`.
 * `AUTH`: Sets the type of the auth service. `fake` (default) or `ticket`.
 * `AUTH_HOST`: Host of the auth service. The default is `localhost`.
 * `AUTH_PORT`: Port of the auth service. The default is `9004`.

cmd/autoupdate/main.go

@@ -48,6 +48,10 @@ func defaultEnv() map[string]string {
 		"MESSAGE_BUS_PORT": "6379",
 		"REDIS_TEST_CONN":  "true",
 
+		"VOTE_HOST":     "localhost",
+		"VOTE_PORT":     "9013",
+		"VOTE_PROTOCAL": "http",
+
 		"AUTH":          "fake",
 		"AUTH_PROTOCOL": "http",
 		"AUTH_HOST":     "localhost",
@@ -135,8 +139,10 @@ func run() error {
 		return fmt.Errorf("creating auth adapter: %w", err)
 	}
 
+	voteAddr := env["VOTE_PROTOCAL"] + "://" + env["VOTE_HOST"] + env["VOTE_PORT"]
+
 	// Autoupdate Service.
-	service := autoupdate.New(datastoreService, restrict.Middleware, ctx.Done())
+	service := autoupdate.New(datastoreService, restrict.Middleware, voteAddr, ctx.Done())
 	go service.PruneOldData(ctx)
 	go service.ResetCache(ctx)
 

internal/autoupdate/autoupdate.go

@@ -44,6 +44,7 @@ type Autoupdate struct {
 	datastore  Datastore
 	topic      *topic.Topic
 	restricter RestrictMiddleware
+	voteAddr   string
 }
 
 // RestrictMiddleware is a function that can restrict data.
@@ -53,11 +54,12 @@ type RestrictMiddleware func(getter datastore.Getter, uid int) datastore.Getter
 //
 // The attribute closed is a channel that should be closed when the server shuts
 // down. In this case, all connections get closed.
-func New(datastore Datastore, restricter RestrictMiddleware, closed <-chan struct{}) *Autoupdate {
+func New(datastore Datastore, restricter RestrictMiddleware, voteAddr string, closed <-chan struct{}) *Autoupdate {
 	a := &Autoupdate{
 		datastore:  datastore,
 		topic:      topic.New(topic.WithClosed(closed)),
 		restricter: restricter,
+		voteAddr:   voteAddr,
 	}
 
 	// Update the topic when an data update is received.
@@ -71,6 +73,9 @@ func New(datastore Datastore, restricter RestrictMiddleware, closed <-chan struc
 		return nil
 	})
 
+	// Register the calculated field for vote_count.
+	a.datastore.RegisterCalculatedField("poll/vote_count", a.datastorePollVoteCount)
+
 	return a
 }
 

internal/autoupdate/connection_test.go

@@ -89,7 +89,7 @@ func TestConnectionEmptyData(t *testing.T) {
 	})
 	go datastore.ListenOnUpdates(shutdownCtx, datastore, func(err error) { log.Println(err) })
 
-	s := autoupdate.New(datastore, test.RestrictAllowed, shutdownCtx.Done())
+	s := autoupdate.New(datastore, test.RestrictAllowed, "", shutdownCtx.Done())
 	kb := test.KeysBuilder{K: test.Str(doesExistKey, doesNotExistKey)}
 
 	t.Run("First responce", func(t *testing.T) {
@@ -202,7 +202,7 @@ func TestConnectionFilterData(t *testing.T) {
 		"user/1/name": []byte(`"Hello World"`),
 	})
 
-	s := autoupdate.New(datastore, test.RestrictAllowed, shutdownCtx.Done())
+	s := autoupdate.New(datastore, test.RestrictAllowed, "", shutdownCtx.Done())
 	kb := test.KeysBuilder{K: test.Str("user/1/name")}
 	next := s.Connect(1, kb)
 	if _, err := next(context.Background()); err != nil {
@@ -233,7 +233,7 @@ func TestConntectionFilterOnlyOneKey(t *testing.T) {
 	})
 	go datastore.ListenOnUpdates(shutdownCtx, datastore, func(err error) { log.Println(err) })
 
-	s := autoupdate.New(datastore, test.RestrictAllowed, shutdownCtx.Done())
+	s := autoupdate.New(datastore, test.RestrictAllowed, "", shutdownCtx.Done())
 	kb := test.KeysBuilder{K: test.Str("user/1/name")}
 	next := s.Connect(1, kb)
 	if _, err := next(context.Background()); err != nil {
@@ -265,7 +265,7 @@ func TestNextNoReturnWhenDataIsRestricted(t *testing.T) {
 		"user/1/name": []byte(`"Hello World"`),
 	})
 
-	s := autoupdate.New(datastore, test.RestrictNotAllowed, shutdownCtx.Done())
+	s := autoupdate.New(datastore, test.RestrictNotAllowed, "", shutdownCtx.Done())
 	kb := test.KeysBuilder{K: test.Str("user/1/name")}
 
 	next := s.Connect(1, kb)
@@ -321,7 +321,7 @@ func TestKeyNotRequestedAnymore(t *testing.T) {
 	`))
 	go datastore.ListenOnUpdates(shutdownCtx, datastore, nil)
 
-	s := autoupdate.New(datastore, test.RestrictAllowed, shutdownCtx.Done())
+	s := autoupdate.New(datastore, test.RestrictAllowed, "", shutdownCtx.Done())
 	kb, err := keysbuilder.FromJSON(strings.NewReader(`{
 		"collection":"organization",
 		"ids":[

internal/autoupdate/feature_test.go

@@ -81,7 +81,7 @@ func TestFeatures(t *testing.T) {
 	defer cancel()
 
 	datastore := dsmock.NewMockDatastore(shutdownCtx.Done(), dsmock.YAMLData(dataSet))
-	service := autoupdate.New(datastore, test.RestrictAllowed, shutdownCtx.Done())
+	service := autoupdate.New(datastore, test.RestrictAllowed, "", shutdownCtx.Done())
 
 	for _, tt := range []struct {
 		name string

internal/autoupdate/interfaces.go

@@ -11,6 +11,8 @@ type Datastore interface {
 	Get(ctx context.Context, keys ...string) (map[string][]byte, error)
 	RegisterChangeListener(f func(map[string][]byte) error)
 	ResetCache()
+	RegisterCalculatedField(field string, f func(ctx context.Context, key string, changed map[string][]byte) ([]byte, error))
+	RequestKeys(url string, keys []string) (map[string][]byte, error)
 }
 
 // KeysBuilder holds the keys that are requested by a user.

internal/autoupdate/mock_test.go

@@ -12,7 +12,7 @@ func getConnection(closed <-chan struct{}) (autoupdate.DataProvider, *dsmock.Moc
 	datastore := dsmock.NewMockDatastore(closed, map[string][]byte{
 		"user/1/name": []byte(`"Hello World"`),
 	})
-	s := autoupdate.New(datastore, test.RestrictAllowed, closed)
+	s := autoupdate.New(datastore, test.RestrictAllowed, "", closed)
 	kb := test.KeysBuilder{K: test.Str("user/1/name")}
 	next := s.Connect(1, kb)
 

internal/autoupdate/poll_vote_count.go

@@ -0,0 +1,20 @@
+package autoupdate
+
+import (
+	"context"
+	"fmt"
+)
+
+const path = "/internal/vote/vote_count"
+
+func (a *Autoupdate) datastorePollVoteCount(ctx context.Context, fqfield string, changed map[string][]byte) ([]byte, error) {
+	if changed != nil {
+		return changed[fqfield], nil
+	}
+
+	values, err := a.datastore.RequestKeys(a.voteAddr+path, []string{fqfield})
+	if err != nil {
+		return nil, fmt.Errorf("loading key %q from vote-service: %w", fqfield, err)
+	}
+	return values[fqfield], nil
+}

pkg/datastore/datastore.go

@@ -181,7 +181,7 @@ func (d *Datastore) ListenOnUpdates(ctx context.Context, keychanger Updater, err
 func (d *Datastore) loadKeys(keys []string, set func(string, []byte)) error {
 	calculatedKeys, normalKeys := d.splitCalculatedKeys(keys)
 	if len(normalKeys) > 0 {
-		data, err := d.requestKeys(normalKeys)
+		data, err := d.RequestKeys(d.url, normalKeys)
 		if err != nil {
 			return fmt.Errorf("requesting keys from datastore: %w", err)
 		}
@@ -218,19 +218,19 @@ func (d *Datastore) calculateField(field string, key string, updated map[string]
 
 }
 
-// requestKeys request a list of keys by the datastore.
+// RequestKeys request a list of keys from the datastore.
 //
 // If an error happens, no key is returned.
 //
 // The returned map contains exacply the given keys. If a key does not exist in
 // the datastore, then the value of this key is <nil>.
-func (d *Datastore) requestKeys(keys []string) (map[string][]byte, error) {
+func (d *Datastore) RequestKeys(url string, keys []string) (map[string][]byte, error) {
 	requestData, err := keysToGetManyRequest(keys)
 	if err != nil {
 		return nil, fmt.Errorf("creating GetManyRequest: %w", err)
 	}
 
-	req, err := http.NewRequest("POST", d.url, bytes.NewReader(requestData))
+	req, err := http.NewRequest("POST", url, bytes.NewReader(requestData))
 	if err != nil {
 		return nil, fmt.Errorf("creating request: %w", err)
 	}