some

Author: tacogips

Diff for: cmd/chain/serve.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"os/signal"
@@ -15,6 +16,7 @@ import (
 	"github.com/ajainc/chain/ctx/logger"
 	chaingrpc "github.com/ajainc/chain/grpc"
 	"github.com/ajainc/chain/grpc/gen"
+	"github.com/improbable-eng/grpc-web/go/grpcweb"
 )
 
 //TODO (tacogips) implement commands with option
@@ -28,24 +30,39 @@ func main() {
 	//  setup grpc server
 	//TODO(taco) move broadcast channel size to config
 	streamBroadcastCh := make(chan *gen.StreamPayload, 300)
-	grpcwebServer, err := chaingrpc.Setup(c, streamBroadcastCh)
+	grpcServer, err := chaingrpc.Setup(c, streamBroadcastCh)
+
 	if err != nil {
 		panic(err)
 	}
 
-	handler := func(resp http.ResponseWriter, req *http.Request) {
-		grpcwebServer.ServeHttp(resp, req)
-	}
+	grpcWeb := false
+	if grpcWeb {
+		grpcwebServer := grpcweb.WrapServer(grpcServer)
 
-	grpcHttpServer := http.Server{
-		Addr:    fmt.Sprintf(":%d", port),
-		Handler: http.HandlerFunc(handler),
-	}
+		handler := func(resp http.ResponseWriter, req *http.Request) {
+			grpcwebServer.ServeHttp(resp, req)
+		}
+		grpcHttpServer := http.Server{
+			Addr:    fmt.Sprintf(":%d", port),
+			Handler: http.HandlerFunc(handler),
+		}
 
-	go func() {
-		logger.Infof(c, "server started. listening on :%d", port)
-		grpcHttpServer.ListenAndServe() // TODO(tacogips): use HTTP/2
-	}()
+		go func() {
+			logger.Infof(c, "grpc web server started. listening on :%d", port)
+			grpcHttpServer.ListenAndServe() // TODO(tacogips): use HTTP/2
+		}()
+
+	} else {
+		listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+		if err != nil {
+			panic(err)
+		}
+		go func() {
+			logger.Infof(c, "grpc server started. listening on :%d", port)
+			grpcServer.Serve(listener)
+		}()
+	}
 
 	waitShutdownDone := make(chan struct{})
 	sigs := make(chan os.Signal, 1)

Diff for: cmd/chaincli/mock.go

@@ -12,12 +12,11 @@ import (
 )
 
 func main() {
-	conn, err := grpc.Dial(":50051")
+	conn, err := grpc.Dial(":50051", grpc.WithInsecure())
 
 	if err != nil {
 		panic(err)
 	}
-
 	defer conn.Close()
 
 	entityClient := gen.NewEntityServiceClient(conn)
@@ -49,7 +48,7 @@ func main() {
 			if err != nil {
 				log.Error(err)
 			}
-			fmt.Printf("%#v", activity)
+			fmt.Printf("%#v\n", activity)
 		}
 	}
 

Diff for: cmd/chainstrmcli/stream-client.go

@@ -10,7 +10,7 @@ import (
 )
 
 func main() {
-	conn, err := grpc.Dial(":50051")
+	conn, err := grpc.Dial(":50051", grpc.WithInsecure())
 
 	if err != nil {
 		panic(err)

Diff for: event/create-entity-ev.go

@@ -8,24 +8,32 @@ import (
 	"github.com/ajainc/chain/ctx/docdb"
 	"github.com/ajainc/chain/event/dao"
 	"github.com/ajainc/chain/grpc/gen"
+	uuid "github.com/satori/go.uuid"
 )
 
 // CeateEntityEv stands for Creating New Entity at specified coordinates
 func NewCeateEntityEvent(entity *gen.Entity) *CeateEntityEv {
 	return &CeateEntityEv{
-		Entity: entity,
+		eventID: uuid.NewV4().String(),
+		Entity:  entity,
 	}
 }
 
 type CeateEntityEv struct {
-	Entity *gen.Entity
+	eventID string
+	Entity  *gen.Entity
+}
+
+func (ev *CeateEntityEv) EventID() string {
+	return ev.eventID
 }
 
 func (ev *CeateEntityEv) Description() string {
 	return fmt.Sprintf("Create Entity")
 }
 
 func (ev *CeateEntityEv) Exec(c context.Context) error {
+
 	db := docdb.FromContext(c)
 
 	ev.Entity.FillWithDefault()

Diff for: event/event.go

@@ -7,6 +7,7 @@ import (
 )
 
 type Event interface {
+	EventID() string
 	Description() string
 	Exec(context.Context) error
 	Undo(context.Context) error

Diff for: event/executor.go

@@ -3,6 +3,7 @@ package event
 import (
 	"context"
 
+	"github.com/ajainc/chain/ctx/logger"
 	"github.com/ajainc/chain/grpc/gen"
 )
 
@@ -11,6 +12,8 @@ func Exec(c context.Context, ev Event, streamBroadcastCh chan *gen.StreamPayload
 
 	var err error
 
+	logger.Debugf(c, "execute event %#v", ev)
+
 	err = ev.Exec(c)
 	if err != nil {
 		return nil, err
@@ -25,6 +28,8 @@ func Exec(c context.Context, ev Event, streamBroadcastCh chan *gen.StreamPayload
 	streamPayloads, _ := ev.ExecStreamPayloads(c)
 	//TODO(taco) retry if error?
 	go func() {
+
+		logger.Debugf(c, "stream out exec result %#v", streamPayloads)
 		for _, paylaod := range streamPayloads {
 			streamBroadcastCh <- paylaod
 		}
@@ -35,12 +40,16 @@ func Exec(c context.Context, ev Event, streamBroadcastCh chan *gen.StreamPayload
 
 //Undo execute ev.Undo()
 func Undo(c context.Context, ev Event, streamBroadcastCh chan *gen.StreamPayload) error {
+
+	logger.Debugf(c, "undo event %#v", ev)
 	// TODO(tacogisp): impl
 	ev.Undo(c)
 
 	streamPayloads, _ := ev.ExecStreamPayloads(c)
 	//TODO(taco) retry if error?
 	go func() {
+
+		logger.Debugf(c, "stream out undo result %#v", streamPayloads)
 		for _, paylaod := range streamPayloads {
 			streamBroadcastCh <- paylaod
 		}

Diff for: grpc/server/entity-server.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 
+	"github.com/ajainc/chain/ctx/logger"
 	"github.com/ajainc/chain/event"
 	"github.com/ajainc/chain/grpc/gen"
 
@@ -15,8 +16,11 @@ type EntityServer struct {
 }
 
 func (server *EntityServer) CreateEntity(_ go16ctx.Context, in *gen.Entity) (*gen.Activity, error) {
+
+	logger.Debug(server.appCtx, "CreatEnity called")
+
 	ev := event.NewCeateEntityEvent(in)
-	activity, err := event.Exec(server.appCtx, ev)
+	activity, err := event.Exec(server.appCtx, ev, server.streamBroadcastCh)
 	if err != nil {
 		return nil, err
 	}

Diff for: grpc/servers.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/ajainc/chain/grpc/gen"
 	"github.com/ajainc/chain/grpc/server"
-	"github.com/improbable-eng/grpc-web/go/grpcweb"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health"
@@ -28,12 +27,10 @@ func registerServices(c context.Context, grpcServer *grpc.Server, streamBroadcas
 	return nil
 }
 
-func Setup(c context.Context, streamBroadcastCh chan *gen.StreamPayload, opts ...grpc.ServerOption) (*grpcweb.WrappedGrpcServer, error) {
+func Setup(c context.Context, streamBroadcastCh chan *gen.StreamPayload, opts ...grpc.ServerOption) (*grpc.Server, error) {
 	s := grpc.NewServer(opts...)
 
 	err := registerServices(c, s, streamBroadcastCh)
 
-	wrappedGrpc := grpcweb.WrapServer(s)
-
-	return wrappedGrpc, err
+	return s, err
 }