grpc

README

Dapr gRPC Service SDK for Go

Start by importing Dapr Go service/grpc package:

daprd "github.com/dapr/go-sdk/service/grpc"

Creating and Starting Service

To create a gRPC Dapr service, first, create a Dapr callback instance with a specific address:

s, err := daprd.NewService(":50001")
if err != nil {
    log.Fatalf("failed to start the server: %v", err)
}

Or with address and an existing net.Listener in case you want to combine existing server listener:

list, err := net.Listen("tcp", "localhost:0")
if err != nil {
	log.Fatalf("gRPC listener creation failed: %s", err)
}
s := daprd.NewServiceWithListener(list)

Dapr gRPC service supports using existed gRPC server with the help of NewServiceWithGrpcServer. You can use RegisterGreeterServer to add existed gRPC service either:

lis, err := net.Listen("tcp", port)
if err != nil {
	log.Fatalf("failed to listen: %v", err)
}

grpcServer := grpc.NewServer()

// register existed service
// pb.RegisterGreeterServer(grpcServer, &existedGrpcServer{})

// new dapr grpc service
s := daprd.NewServiceWithGrpcServer(lis, grpcServer)

Once you create a service instance, you can "attach" to that service any number of event, binding, and service invocation logic handlers as shown below. Onces the logic is defined, you are ready to start the service:

if err := s.Start(); err != nil {
    log.Fatalf("server error: %v", err)
}

Event Handling

To handle events from specific topic you need to add at least one topic event handler before starting the service:

sub := &common.Subscription{
		PubsubName: "messages",
		Topic:      "topic1",
	}
if err := s.AddTopicEventHandler(sub, eventHandler); err != nil {
    log.Fatalf("error adding topic subscription: %v", err)
}

The handler method itself can be any method with the expected signature:

func eventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err error) {
	log.Printf("event - PubsubName:%s, Topic:%s, ID:%s, Data: %v", e.PubsubName, e.Topic, e.ID, e.Data)
	// do something with the event
	return true, nil
}

Service Invocation Handler

To handle service invocations you will need to add at least one service invocation handler before starting the service:

if err := s.AddServiceInvocationHandler("echo", echoHandler); err != nil {
    log.Fatalf("error adding invocation handler: %v", err)
}

The handler method itself can be any method with the expected signature:

func echoHandler(ctx context.Context, in *common.InvocationEvent) (out *common.Content, err error) {
	log.Printf("echo - ContentType:%s, Verb:%s, QueryString:%s, %+v", in.ContentType, in.Verb, in.QueryString, string(in.Data))
	// do something with the invocation here
	out = &common.Content{
		Data:        in.Data,
		ContentType: in.ContentType,
		DataTypeURL: in.DataTypeURL,
	}
	return
}

Binding Invocation Handler

To handle binding invocations you will need to add at least one binding invocation handler before starting the service:

if err := s.AddBindingInvocationHandler("run", runHandler); err != nil {
    log.Fatalf("error adding binding handler: %v", err)
}

The handler method itself can be any method with the expected signature:

func runHandler(ctx context.Context, in *common.BindingEvent) (out []byte, err error) {
	log.Printf("binding - Data:%v, Meta:%v", in.Data, in.Metadata)
	// do something with the invocation here
	return nil, nil
}

Templates

To accelerate your gRPC Dapr app development in Go even further you can use one of the GitHub templates integrating the gRPC Dapr callback package:

• Dapr gRPC Service in Go - Template project to jump start your Dapr event subscriber service with gRPC development
• Dapr gRPC Event Subscriber in Go - Template project to jump start your Dapr event subscriber service with gRPC development

Contributing to Dapr Go client

See the Contribution Guide to get started with building and developing.

Documentation

Index

• func NewService(address string) (s common.Service, err error)
• func NewServiceWithGrpcServer(lis net.Listener, server *grpc.Server) common.Service
• func NewServiceWithListener(lis net.Listener, opts ...grpc.ServerOption) common.Service
• type Server
  • func (s *Server) AddBindingInvocationHandler(name string, fn common.BindingInvocationHandler) error
  • func (s *Server) AddHealthCheckHandler(_ string, fn common.HealthCheckHandler) error
  • func (s *Server) AddServiceInvocationHandler(method string, fn cc.ServiceInvocationHandler) error
  • func (s *Server) AddTopicEventHandler(sub *common.Subscription, fn common.TopicEventHandler) error
  • func (s *Server) GracefulStop() error
  • func (s *Server) GrpcServer() *grpc.Server
  • func (s *Server) HealthCheck(ctx context.Context, _ *emptypb.Empty) (*pb.HealthCheckResponse, error)
  • func (s *Server) ListInputBindings(ctx context.Context, in *emptypb.Empty) (*pb.ListInputBindingsResponse, error)
  • func (s *Server) ListTopicSubscriptions(ctx context.Context, in *emptypb.Empty) (*runtimev1pb.ListTopicSubscriptionsResponse, error)
  • func (s *Server) OnBindingEvent(ctx context.Context, in *pb.BindingEventRequest) (*pb.BindingEventResponse, error)
  • func (s *Server) OnInvoke(ctx context.Context, in *cpb.InvokeRequest) (*cpb.InvokeResponse, error)
  • func (s *Server) OnTopicEvent(ctx context.Context, in *runtimev1pb.TopicEventRequest) (*runtimev1pb.TopicEventResponse, error)
  • func (s *Server) RegisterActorImplFactory(f actor.Factory, opts ...config.Option)
  • func (s *Server) RegisterActorImplFactoryContext(f actor.FactoryContext, opts ...config.Option)
  • func (s *Server) Start() error
  • func (s *Server) Stop() error

Functions

func NewService

func NewService(address string) (s common.Service, err error)

NewService creates new Service.

func NewServiceWithGrpcServer

func NewServiceWithGrpcServer(lis net.Listener, server *grpc.Server) common.Service

NewServiceWithGrpcServer creates a new Service with specific listener and grpcServer

func NewServiceWithListener

func NewServiceWithListener(lis net.Listener, opts ...grpc.ServerOption) common.Service

NewServiceWithListener creates new Service with specific listener.

Types

type Server

type Server struct {
	pb.UnimplementedAppCallbackServer
	pb.UnimplementedAppCallbackHealthCheckServer
	// contains filtered or unexported fields
}

Server is the gRPC service implementation for Dapr.

func (*Server) AddBindingInvocationHandler

func (s *Server) AddBindingInvocationHandler(name string, fn common.BindingInvocationHandler) error

AddBindingInvocationHandler appends provided binding invocation handler with its name to the service.

func (*Server) AddHealthCheckHandler

func (s *Server) AddHealthCheckHandler(_ string, fn common.HealthCheckHandler) error

AddHealthCheckHandler appends provided app health check handler.

func (*Server) AddServiceInvocationHandler

func (s *Server) AddServiceInvocationHandler(method string, fn cc.ServiceInvocationHandler) error

AddServiceInvocationHandler appends provided service invocation handler with its method to the service.

func (*Server) AddTopicEventHandler

func (s *Server) AddTopicEventHandler(sub *common.Subscription, fn common.TopicEventHandler) error

AddTopicEventHandler appends provided event handler with topic name to the service.

func (*Server) GracefulStop

func (s *Server) GracefulStop() error

GrecefulStop stops the previously-started service gracefully.

func (*Server) GrpcServer

func (s *Server) GrpcServer() *grpc.Server

GrpcServer returns the grpc.Server object managed by the server.

func (*Server) HealthCheck

func (s *Server) HealthCheck(ctx context.Context, _ *emptypb.Empty) (*pb.HealthCheckResponse, error)

HealthCheck check app health status.

func (*Server) ListInputBindings

func (s *Server) ListInputBindings(ctx context.Context, in *emptypb.Empty) (*pb.ListInputBindingsResponse, error)

ListInputBindings is called by Dapr to get the list of bindings the app will get invoked by. In this example, we are telling Dapr To invoke our app with a binding named storage.

func (*Server) ListTopicSubscriptions

func (s *Server) ListTopicSubscriptions(ctx context.Context, in *emptypb.Empty) (*runtimev1pb.ListTopicSubscriptionsResponse, error)

ListTopicSubscriptions is called by Dapr to get the list of topics in a pubsub component the app wants to subscribe to.

func (*Server) OnBindingEvent

func (s *Server) OnBindingEvent(ctx context.Context, in *pb.BindingEventRequest) (*pb.BindingEventResponse, error)

OnBindingEvent gets invoked every time a new event is fired from a registered binding. The message carries the binding name, a payload and optional metadata.

func (*Server) OnInvoke

func (s *Server) OnInvoke(ctx context.Context, in *cpb.InvokeRequest) (*cpb.InvokeResponse, error)

OnInvoke gets invoked when a remote service has called the app through Dapr.

func (*Server) OnTopicEvent

func (s *Server) OnTopicEvent(ctx context.Context, in *runtimev1pb.TopicEventRequest) (*runtimev1pb.TopicEventResponse, error)

OnTopicEvent fired whenever a message has been published to a topic that has been subscribed. Dapr sends published messages in a CloudEvents v1.0 envelope.

func (*Server) RegisterActorImplFactory

func (s *Server) RegisterActorImplFactory(f actor.Factory, opts ...config.Option)

Deprecated: Use RegisterActorImplFactoryContext instead.

func (*Server) RegisterActorImplFactoryContext

func (s *Server) RegisterActorImplFactoryContext(f actor.FactoryContext, opts ...config.Option)

func (*Server) Start

func (s *Server) Start() error

Start registers the server and starts it.

func (*Server) Stop

func (s *Server) Stop() error

Stop stops the previously-started service.