protonats

README

Protobuf for Nats.io

This Go module simplifies how you use protobuf with Nats.io:

1. Provide proto files
2. Provide Go implementations
3. Bind the implementation on the server side
4. Make a call from the clients

Example

Proto:

message TestARequest {
    string input = 1; 
}

message TestAResponse {
    string output = 1;
}

service TestService {
    rpc GetTestA(TestARequest) returns (TestAResponse) {}
}

Implementation:

func (b *TestServiceProtonats) GetTestA(ctx context.Context, req *TestARequest) (*TestAResponse, error) {
	...
    //some business
    ...
    if err != nil {
		return nil, err
	}

	result := &TestAResponse{
		Output: "OK",
	}
	return result, nil
}

Server side:

Generate the server code from the proto file with --protonats_out= argument to protoc-gogo, then:

    service := &TestServiceProtonats{}
    ...
    ctx, cancel := context.WithCancel(context.Background())
	bus, err := NewBus(ctx, ServiceConfiguration{URL: natsURL, ID: "bus1"})
	defer bus.Close()
	svr := NewTestServiceProtonatsServer(bus, d)
	chan, err = svr.SubscribeTestService()

    ...
	cancel() // cancel services
	...
	<-chan  // wait and unsubscribe 

Client side:

Generate the client code from the proto file with --protonats_out= argument to protoc-gogo, then:

    var client *Bus
	client, err = NewBus(ctx, ServiceConfiguration{URL: natsURL})
	
	defer client.Close()

	svc := NewTestServiceProtonatsClient(client)
	resp, err := svc.GetTestA(ctx, &TestARequest{Input: "OK"})

GRPC Gateway

A GRPC gateway listens GRPC requests and forward them to NATS. The gateway can be generated by generating code from the proto file with --protonats_out=gprc: argument to protoc-gogo. See grpc_test.go file to check how toimplement the bridge.

License

This software is licensed under Apache 2 license.

(c) 2019 Citra Digital Lintas

Documentation

Index

• Variables
• type Bus
  • func NewBus(ctx context.Context, config ServiceConfiguration) (*Bus, error)
  • func (bus *Bus) Close()
  • func (bus *Bus) HandleError(replySubject string, err error)
• type ErrorMessage
  • func (*ErrorMessage) Descriptor() ([]byte, []int)
  • func (m *ErrorMessage) GetBusID() string
  • func (m *ErrorMessage) GetErrorMessage() string
  • func (m *ErrorMessage) GetTimestamp() int64
  • func (m *ErrorMessage) Marshal() (dAtA []byte, err error)
  • func (m *ErrorMessage) MarshalTo(dAtA []byte) (int, error)
  • func (m *ErrorMessage) MarshalToSizedBuffer(dAtA []byte) (int, error)
  • func (*ErrorMessage) ProtoMessage()
  • func (m *ErrorMessage) Reset()
  • func (m *ErrorMessage) Size() (n int)
  • func (m *ErrorMessage) String() string
  • func (m *ErrorMessage) Unmarshal(dAtA []byte) error
  • func (m *ErrorMessage) XXX_DiscardUnknown()
  • func (m *ErrorMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
  • func (m *ErrorMessage) XXX_Merge(src proto.Message)
  • func (m *ErrorMessage) XXX_Size() int
  • func (m *ErrorMessage) XXX_Unmarshal(b []byte) error
• type ServiceConfiguration
• type StreamInfo
  • func (*StreamInfo) Descriptor() ([]byte, []int)
  • func (m *StreamInfo) GetID() string
  • func (m *StreamInfo) Marshal() (dAtA []byte, err error)
  • func (m *StreamInfo) MarshalTo(dAtA []byte) (int, error)
  • func (m *StreamInfo) MarshalToSizedBuffer(dAtA []byte) (int, error)
  • func (*StreamInfo) ProtoMessage()
  • func (m *StreamInfo) Reset()
  • func (m *StreamInfo) Size() (n int)
  • func (m *StreamInfo) String() string
  • func (m *StreamInfo) Unmarshal(dAtA []byte) error
  • func (m *StreamInfo) XXX_DiscardUnknown()
  • func (m *StreamInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
  • func (m *StreamInfo) XXX_Merge(src proto.Message)
  • func (m *StreamInfo) XXX_Size() int
  • func (m *StreamInfo) XXX_Unmarshal(b []byte) error

Variables

var (
	ErrInvalidLengthNats = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowNats   = fmt.Errorf("proto: integer overflow")
)

Types

type Bus

type Bus struct {
	Connection    *nats.Conn
	Configuration ServiceConfiguration
	Context       context.Context
}

func NewBus

func NewBus(ctx context.Context, config ServiceConfiguration) (*Bus, error)

func (*Bus) Close

func (bus *Bus) Close()

func (*Bus) HandleError

func (bus *Bus) HandleError(replySubject string, err error)

type ErrorMessage

type ErrorMessage struct {
	ErrorMessage string `protobuf:"bytes,1,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"`
	Timestamp    int64  `protobuf:"varint,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
	BusID        string `protobuf:"bytes,3,opt,name=busID,proto3" json:"busID,omitempty"`
}

func (*ErrorMessage) Descriptor

func (*ErrorMessage) Descriptor() ([]byte, []int)

func (*ErrorMessage) GetBusID

func (m *ErrorMessage) GetBusID() string

func (*ErrorMessage) GetErrorMessage

func (m *ErrorMessage) GetErrorMessage() string

func (*ErrorMessage) GetTimestamp

func (m *ErrorMessage) GetTimestamp() int64

func (*ErrorMessage) Marshal

func (m *ErrorMessage) Marshal() (dAtA []byte, err error)

func (*ErrorMessage) MarshalTo

func (m *ErrorMessage) MarshalTo(dAtA []byte) (int, error)

func (*ErrorMessage) MarshalToSizedBuffer

func (m *ErrorMessage) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*ErrorMessage) ProtoMessage

func (*ErrorMessage) ProtoMessage()

func (*ErrorMessage) Reset

func (m *ErrorMessage) Reset()

func (*ErrorMessage) Size

func (m *ErrorMessage) Size() (n int)

func (*ErrorMessage) String

func (m *ErrorMessage) String() string

func (*ErrorMessage) Unmarshal

func (m *ErrorMessage) Unmarshal(dAtA []byte) error

func (*ErrorMessage) XXX_DiscardUnknown

func (m *ErrorMessage) XXX_DiscardUnknown()

func (*ErrorMessage) XXX_Marshal

func (m *ErrorMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*ErrorMessage) XXX_Merge

func (m *ErrorMessage) XXX_Merge(src proto.Message)

func (*ErrorMessage) XXX_Size

func (m *ErrorMessage) XXX_Size() int

func (*ErrorMessage) XXX_Unmarshal

func (m *ErrorMessage) XXX_Unmarshal(b []byte) error

type ServiceConfiguration

type ServiceConfiguration struct {
	URL string
	ID  string
}

type StreamInfo

type StreamInfo struct {
	ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
}

func (*StreamInfo) Descriptor

func (*StreamInfo) Descriptor() ([]byte, []int)

func (*StreamInfo) GetID

func (m *StreamInfo) GetID() string

func (*StreamInfo) Marshal

func (m *StreamInfo) Marshal() (dAtA []byte, err error)

func (*StreamInfo) MarshalTo

func (m *StreamInfo) MarshalTo(dAtA []byte) (int, error)

func (*StreamInfo) MarshalToSizedBuffer

func (m *StreamInfo) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*StreamInfo) ProtoMessage

func (*StreamInfo) ProtoMessage()

func (*StreamInfo) Reset

func (m *StreamInfo) Reset()

func (*StreamInfo) Size

func (m *StreamInfo) Size() (n int)

func (*StreamInfo) String

func (m *StreamInfo) String() string

func (*StreamInfo) Unmarshal

func (m *StreamInfo) Unmarshal(dAtA []byte) error

func (*StreamInfo) XXX_DiscardUnknown

func (m *StreamInfo) XXX_DiscardUnknown()

func (*StreamInfo) XXX_Marshal

func (m *StreamInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*StreamInfo) XXX_Merge

func (m *StreamInfo) XXX_Merge(src proto.Message)

func (*StreamInfo) XXX_Size

func (m *StreamInfo) XXX_Size() int

func (*StreamInfo) XXX_Unmarshal

func (m *StreamInfo) XXX_Unmarshal(b []byte) error