some

Author: kawada_yusuke

event/add-relation-ev.go

@@ -64,6 +64,7 @@ func (ev *AddRelationEv) ExecStreamPayloads(c context.Context) ([]*gen.StreamPay
 		return nil, err
 	}
 
+	//TODO(taco) use helper
 	return []*gen.StreamPayload{
 		{
 			Operation: gen.StreamPayload_NEW,
@@ -87,6 +88,7 @@ func (ev *AddRelationEv) Undo(c context.Context) error {
 }
 
 func (ev *AddRelationEv) UndoStreamPayloads(c context.Context) ([]*gen.StreamPayload, error) {
+	//TODO(taco) use helper
 	return []*gen.StreamPayload{
 		{
 			Operation: gen.StreamPayload_DELETE,

event/dao/entity.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/ajainc/chain/ctx/docdb"
+	"github.com/ajainc/chain/ctx/logger"
 	"github.com/ajainc/chain/grpc/gen"
 	"github.com/fatih/structs"
 	"github.com/mitchellh/mapstructure"
@@ -63,6 +64,33 @@ func GetEntityByObjectID(c context.Context, objectID string) (int, gen.Entity, e
 	return id, *entity, nil
 }
 
+//GetAllEntities
+func GetAllEntities(c context.Context) ([]*gen.Entity, error) {
+	db := docdb.FromContext(c)
+
+	entities := db.Use(docdb.COLL_ENTITY)
+	var result []*gen.Entity
+	entities.ForEachDoc(func(id int, docContent []byte) bool {
+		doc, err := entities.Read(id)
+		if err != nil {
+			logger.Error(c, err)
+			return false
+		}
+
+		entity := new(gen.Entity)
+		err = UnmarshalEntity(doc, entity)
+		if err != nil {
+			logger.Error(c, err)
+			return false
+		}
+		result = append(result, entity)
+
+		return true
+	})
+
+	return result, nil
+}
+
 func UnmarshalEntity(src map[string]interface{}, dst *gen.Entity) error {
 	return mapstructure.Decode(src, dst)
 }

event/dao/relation.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/ajainc/chain/ctx/docdb"
+	"github.com/ajainc/chain/ctx/logger"
 	"github.com/ajainc/chain/grpc/gen"
 	"github.com/fatih/structs"
 	"github.com/mitchellh/mapstructure"
@@ -44,6 +45,33 @@ func GetRelationByObjectID(c context.Context, objectID string) (int, gen.Rel, er
 	return id, *rel, nil
 }
 
+//GetAllRelations
+func GetAllRelations(c context.Context) ([]*gen.Rel, error) {
+	db := docdb.FromContext(c)
+
+	rels := db.Use(docdb.COLL_REL)
+	var result []*gen.Rel
+	rels.ForEachDoc(func(id int, docContent []byte) bool {
+		doc, err := rels.Read(id)
+		if err != nil {
+			logger.Error(c, err)
+			return false
+		}
+
+		entity := new(gen.Rel)
+		err = UnmarshalRelation(doc, entity)
+		if err != nil {
+			logger.Error(c, err)
+			return false
+		}
+		result = append(result, entity)
+
+		return true
+	})
+
+	return result, nil
+}
+
 func UnmarshalRelation(src map[string]interface{}, dst *gen.Rel) error {
 	return mapstructure.Decode(src, dst)
 }

event/executor.go

@@ -51,7 +51,6 @@ func Undo(c context.Context, ev Event, streamBroadcastCh chan *gen.StreamPayload
 	streamPayloads, _ := ev.UndoStreamPayloads(c)
 	//TODO(taco) retry if error?
 	go func() {
-
 		logger.Debugf(c, "stream out undo result %#v", streamPayloads)
 		for _, paylaod := range streamPayloads {
 			streamBroadcastCh <- paylaod

grpc/gen/stream-helper.go

@@ -0,0 +1,19 @@
+package gen
+
+func NewEntitySteamPayload(ope StreamPayload_Operation, entity *Entity) *StreamPayload {
+	return &StreamPayload{
+		Operation: ope,
+		Object: &StreamPayload_Entity{
+			entity,
+		},
+	}
+}
+
+func NewRelationSteamPayload(ope StreamPayload_Operation, rel *Rel) *StreamPayload {
+	return &StreamPayload{
+		Operation: ope,
+		Object: &StreamPayload_Rel{
+			rel,
+		},
+	}
+}

grpc/server/stream-server.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"github.com/ajainc/chain/ctx/logger"
+	"github.com/ajainc/chain/event/dao"
 	"github.com/ajainc/chain/grpc/gen"
 )
 
@@ -81,6 +82,26 @@ func (server *StreamServer) Connect(req *gen.StreamConnectReq, stream gen.Stream
 			return err
 		}
 		server.addListener(newListener)
+
+		// load all data
+		//TODO (taco) move to apropriate place
+		go func() {
+			entities, err := dao.GetAllEntities(server.appCtx)
+			if err != nil {
+				logger.Errorf(server.appCtx, "failed to get all entities before send to stream")
+			}
+			for _, entity := range entities {
+				newListener.Send(gen.NewEntitySteamPayload(gen.StreamPayload_NEW, entity))
+			}
+
+			rels, err := dao.GetAllRelations(server.appCtx)
+
+			for _, rel := range rels {
+				newListener.Send(gen.NewRelationSteamPayload(gen.StreamPayload_NEW, rel))
+			}
+
+		}()
+
 		err = newListener.Listen() // wait
 		if err != nil {
 			logger.Warnf(server.appCtx, err.Error())