[issue_63] implemented eventing support. TCK pass. Minor fixes. (#64)

cloudstateio · Dec 10, 2020 · 11a84a3 · 11a84a3
1 parent 25bd0ff
commit 11a84a3
Show file tree

Hide file tree

Showing 33 changed files with 3,530 additions and 1,054 deletions.
diff --git a/build/compile-pb.sh b/build/compile-pb.sh
@@ -8,6 +8,11 @@ protoc --go-grpc_out=paths=source_relative:cloudstate/protocol --proto_path=prot
 protoc --go_out=paths=source_relative:cloudstate/protocol --proto_path=protobuf/protocol/cloudstate entity.proto
 protoc --go-grpc_out=paths=source_relative:. --proto_path=protobuf/frontend/ cloudstate/entity_key.proto
 protoc --go_out=paths=source_relative:. --proto_path=protobuf/frontend/ cloudstate/entity_key.proto
+protoc --go-grpc_out=paths=source_relative:. --proto_path=protobuf/frontend/ cloudstate/legacy_entity_key.proto
+protoc --go_out=paths=source_relative:. --proto_path=protobuf/frontend/ cloudstate/legacy_entity_key.proto
+protoc --go-grpc_out=paths=source_relative:. --proto_path=protobuf/frontend/ cloudstate/eventing.proto
+protoc --go_out=paths=source_relative:. --proto_path=protobuf/frontend/ cloudstate/eventing.proto
+
 protoc --go-grpc_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol --proto_path=protobuf/protocol/cloudstate crdt.proto
 protoc --go_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol --proto_path=protobuf/protocol/cloudstate crdt.proto
 protoc --go-grpc_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol/ --proto_path=protobuf/protocol/cloudstate event_sourced.proto
@@ -17,6 +22,9 @@ protoc --go_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/pr
 protoc --go-grpc_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol/ --proto_path=protobuf/protocol/cloudstate value_entity.proto
 protoc --go_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol/ --proto_path=protobuf/protocol/cloudstate value_entity.proto
 
+protoc --go-grpc_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol/ --proto_path=protobuf/protocol/cloudstate value_entity.proto
+protoc --go_out=paths=source_relative:cloudstate/entity --proto_path=protobuf/protocol/ --proto_path=protobuf/protocol/cloudstate value_entity.proto
+
 # TCK CRDT
 protoc --go-grpc_out=paths=source_relative:./tck/crdt \
   --proto_path=protobuf/protocol \
@@ -61,6 +69,22 @@ protoc --go_out=paths=source_relative:./tck/eventsourced \
   --proto_path=protobuf/tck/cloudstate/tck/model \
   --proto_path=protobuf/tck eventsourced.proto
 
+# TCK Eventlog Eventing
+protoc --go-grpc_out=paths=source_relative:./tck/eventlogeventing \
+  --proto_path=protobuf/protocol \
+  --proto_path=protobuf/frontend \
+  --proto_path=protobuf/frontend/cloudstate \
+  --proto_path=protobuf/proxy \
+  --proto_path=protobuf/tck/cloudstate/tck/model \
+  --proto_path=protobuf/tck eventlogeventing.proto
+protoc --go_out=paths=source_relative:./tck/eventlogeventing \
+  --proto_path=protobuf/protocol \
+  --proto_path=protobuf/frontend \
+  --proto_path=protobuf/frontend/cloudstate \
+  --proto_path=protobuf/proxy \
+  --proto_path=protobuf/tck/cloudstate/tck/model \
+  --proto_path=protobuf/tck eventlogeventing.proto
+
 # TCK Action
 protoc --go-grpc_out=paths=source_relative:./tck/action \
   --proto_path=protobuf/protocol \

diff --git a/cloudstate/action/context.go b/cloudstate/action/context.go
@@ -53,9 +53,9 @@ type Context struct {
 	cancelled bool
 }
 
-func (c *Context) RespondWith(reply *any.Any) {
+func (c *Context) RespondWith(response *any.Any) {
 	c.failure = nil
-	c.response = reply
+	c.response = response
 	c.forward = nil
 }
 

diff --git a/cloudstate/action/server.go b/cloudstate/action/server.go
@@ -314,6 +314,9 @@ type runner struct {
 func (r *runner) runCommand(cmd *entity.ActionCommand) error {
 	// unmarshal the commands message
 	msgName := strings.TrimPrefix(cmd.GetPayload().GetTypeUrl(), "type.googleapis.com/")
+	if strings.HasPrefix(msgName, "json.cloudstate.io/") {
+		return r.context.Instance.HandleCommand(r.context, cmd.Name, cmd.Payload)
+	}
 	messageType := proto.MessageType(msgName)
 	message, ok := reflect.New(messageType.Elem()).Interface().(proto.Message)
 	if !ok {

diff --git a/cloudstate/cloudstate.go b/cloudstate/cloudstate.go
@@ -82,7 +82,7 @@ func (cs *CloudState) RegisterCRDT(entity *crdt.Entity, config protocol.Descript
 	return nil
 }
 
-// RegisterCRDT registers a CRDT entity.
+// RegisterAction registers an action entity.
 func (cs *CloudState) RegisterAction(entity *action.Entity, config protocol.DescriptorConfig) error {
 	if err := cs.actionServer.Register(entity); err != nil {
 		return err

diff --git a/cloudstate/crdt/cmdcontext.go b/cloudstate/crdt/cmdcontext.go
@@ -112,6 +112,9 @@ func (c *CommandContext) WriteConsistency(wc entity.CrdtWriteConsistency) {
 func (c *CommandContext) runCommand(cmd *protocol.Command) (*any.Any, error) {
 	// unmarshal the commands message
 	msgName := strings.TrimPrefix(cmd.GetPayload().GetTypeUrl(), "type.googleapis.com/")
+	if strings.HasPrefix(msgName, "json.cloudstate.io/") {
+		return c.Instance.HandleCommand(c, cmd.Name, cmd.Payload)
+	}
 	messageType := proto.MessageType(msgName)
 	message, ok := reflect.New(messageType.Elem()).Interface().(proto.Message)
 	if !ok {

diff --git a/cloudstate/entity_key.pb.go b/cloudstate/entity_key.pb.go