Write docs

huangjunwen · huangjunwen · commit 317977359a62 · 2020-07-31T07:43:30.000+08:00
diff --git a/binlogmsg/binlogmsg.go b/binlogmsg/binlogmsg.go
@@ -21,6 +21,9 @@ import (
 	nppbmsg "github.com/huangjunwen/nproto/v2/pb/msg"
 )
 
+// MsgPipe pipes messages from MySQL (>=8.0.2) msg tables to downstream.
+// Messages from MsgPipe have type *rawenc.RawData. So downstream must be able
+// to handle this type of messages.
 type MsgPipe struct {
 	// Immutable fields.
 	downstream  interface{} // msg.MsgPublisher or msg.MsgAsyncPublisher
@@ -36,8 +39,14 @@ type MsgPipe struct {
 // MsgTableFilter returns true if a given table is a msg table.
 type MsgTableFilter func(schema, table string) bool
 
+// MsgPipeOption is option in creating MsgPipe.
 type MsgPipeOption func(*MsgPipe) error
 
+// NewMsgPipe creates a new msg pipe:
+//   - downstream: must be MsgPublisher or MsgAsyncPublisher.
+//   - masterCfg: master connection to read (full dump) and write (delete published messages).
+//   - slaveCfg: slave config for binlog subscription.
+//   - tableFilter: determine whether a table is used to store messages.
 func NewMsgPipe(
 	downstream interface{},
 	masterCfg *mycanal.FullDumpConfig,
@@ -71,6 +80,7 @@ func NewMsgPipe(
 	return pipe, nil
 }
 
+// Run the main loop (flush messages to downstream) until context.Done().
 func (pipe *MsgPipe) Run(ctx context.Context) (err error) {
 	for {
 		pipe.run(ctx)
@@ -309,6 +319,11 @@ func (pipe *MsgPipe) flushMsgEntry(ctx context.Context, entry msgEntry, cb func(
 
 }
 
+// NewMsgPublisher creates a publisher to publish (store) message to MySQL msg tables:
+//   - encoder: encoder for messages.
+//   - q: *sql.DB/*sql.Tx/*sql.Conn/...
+//   - schema: database name.
+//   - table: msg table name, the table must be created by CreateMsgTable.
 func NewMsgPublisher(encoder npenc.Encoder, q sqlh.Queryer, schema, table string) MsgPublisherFunc {
 	return func(ctx context.Context, spec MsgSpec, msg interface{}) error {
 		if err := AssertMsgType(spec, msg); err != nil {
diff --git a/binlogmsg/doc.go b/binlogmsg/doc.go
@@ -0,0 +1,5 @@
+// Package binlogmsg contains publisher implemenation to 'publish' (store) messages to MySQL8
+// tables then flush to downstream publisher using binlog notification.
+//
+// Since messages are stored in normal MySQL tables, all ACID properties are applied to them.
+package binlogmsg
diff --git a/binlogmsg/options.go b/binlogmsg/options.go
@@ -8,13 +8,17 @@ import (
 )
 
 var (
+	// DefaultLockName is the default value of PipeOptLockName.
 	DefaultLockName = "nproto.binlogmsg"
 
+	// DefaultMaxInflight is the default value of PipeOptMaxInflight.
 	DefaultMaxInflight = 4096
 
+	// DefaultRetryWait is the default value of PipeOptRetryWait.
 	DefaultRetryWait = 5 * time.Second
 )
 
+// PipeOptLogger sets logger for MsgPipe.
 func PipeOptLogger(logger logr.Logger) MsgPipeOption {
 	return func(pipe *MsgPipe) error {
 		if logger == nil {
@@ -25,6 +29,8 @@ func PipeOptLogger(logger logr.Logger) MsgPipeOption {
 	}
 }
 
+// PipeOptLockName sets the lock name using in MySQL get lock ("SELECT GET_LOCK"):
+// only one instance of pipes can run for the same lock name.
 func PipeOptLockName(lockName string) MsgPipeOption {
 	return func(pipe *MsgPipe) error {
 		if lockName == "" {
@@ -35,6 +41,7 @@ func PipeOptLockName(lockName string) MsgPipeOption {
 	}
 }
 
+// PipeOptMaxInflight sets the max number of messages inflight (publishing).
 func PipeOptMaxInflight(maxInflight int) MsgPipeOption {
 	return func(pipe *MsgPipe) error {
 		if maxInflight < 1 {
@@ -45,6 +52,7 @@ func PipeOptMaxInflight(maxInflight int) MsgPipeOption {
 	}
 }
 
+// PipeOptRetryWait sets the interval between retries due to all kinds of errors.
 func PipeOptRetryWait(t time.Duration) MsgPipeOption {
 	return func(pipe *MsgPipe) error {
 		if t <= 0 {
diff --git a/binlogmsg/pj.go b/binlogmsg/pj.go
@@ -10,6 +10,9 @@ import (
 	. "github.com/huangjunwen/nproto/v2/msg"
 )
 
+// PbJsonPublisher creates a msg publisher using protobuf or json for encoding:
+//   - If msg is proto.Message, then use protobuf.
+//   - Otherwise use json.
 func PbJsonPublisher(q sqlh.Queryer, schema, table string) MsgPublisherFunc {
 
 	pbPublisher := NewMsgPublisher(pjenc.DefaultPbEncoder, q, schema, table)
diff --git a/msg/common.go b/msg/common.go
@@ -1,4 +1,3 @@
-// Package msg contains high level types/interfaces for msg implementations.
 package msg
 
 import (
@@ -98,6 +97,7 @@ func (spec *msgSpec) String() string {
 	return fmt.Sprintf("MsgSpec(%s %s)", spec.subjectName, spec.msgType.String())
 }
 
+// MustRawDataMsgSpec is must-version of NewRawDataMsgSpec.
 func MustRawDataMsgSpec(subjectName string) MsgSpec {
 	spec, err := NewRawDataMsgSpec(subjectName)
 	if err != nil {
@@ -106,6 +106,7 @@ func MustRawDataMsgSpec(subjectName string) MsgSpec {
 	return spec
 }
 
+// NewRawDataMsgSpec validates and creates a new MsgSpec with *rawenc.RawData msg type.
 func NewRawDataMsgSpec(subjectName string) (MsgSpec, error) {
 	if !SubjectNameRegexp.MatchString(subjectName) {
 		return nil, fmt.Errorf("SubjectName format invalid")
diff --git a/msg/doc.go b/msg/doc.go
@@ -0,0 +1,2 @@
+// Package msg contains high level types/interfaces for msg implementations.
+package msg
diff --git a/msg/publisher.go b/msg/publisher.go
@@ -4,7 +4,7 @@ import (
 	"context"
 )
 
-// MsgPublisher is used to publish messages reliably, e.g. at least once delivery.
+// MsgPublisher is used to publish messages.
 type MsgPublisher interface {
 	// Publish publishes a message to the given subject. It returns nil if success.
 	Publish(ctx context.Context, spec MsgSpec, msg interface{}) error
diff --git a/msg/subscriber.go b/msg/subscriber.go
@@ -14,8 +14,9 @@ type MsgSubscriber interface {
 	Subscribe(spec MsgSpec, queue string, handler MsgHandler, opts ...interface{}) error
 }
 
-// MsgHandler handles messages. A message should be redelivered if the handler returns an error.
-// Except that the error's type is nproto.Error and error.Retryable() is false.
+// MsgHandler handles messages. For 'at least once delivery' implementations, a message
+// should be redelivered if the handler returns an error. Otherwise it may or may not
+// be redelivered.
 type MsgHandler func(context.Context, interface{}) error
 
 // MsgSubscriberFunc is an adapter to allow the use of ordinary functions as MsgSubscriber.
diff --git a/natsrpc/client.go b/natsrpc/client.go
@@ -16,15 +16,19 @@ import (
 	. "github.com/huangjunwen/nproto/v2/rpc"
 )
 
+// ClientConn wraps nats.Conn into 'client side rpc connection'.
 type ClientConn struct {
 	// Immutable fields.
 	subjectPrefix string
 	timeout       time.Duration
 	nc            *nats.Conn
 }
 
+// ClientConnOption is option in creating ClientConn.
 type ClientConnOption func(*ClientConn) error
 
+// NewClientConn creates a new ClientConn. `nc` must have MaxReconnect < 0
+// (e.g. never give up trying to reconnect).
 func NewClientConn(nc *nats.Conn, opts ...ClientConnOption) (*ClientConn, error) {
 
 	if nc.Opts.MaxReconnect >= 0 {
@@ -45,6 +49,7 @@ func NewClientConn(nc *nats.Conn, opts ...ClientConnOption) (*ClientConn, error)
 	return cc, nil
 }
 
+// Client creates an rpc client using specified encoder and decoder.
 func (cc *ClientConn) Client(encoder npenc.Encoder, decoder npenc.Decoder) RPCClientFunc {
 	return func(spec RPCSpec) RPCHandler {
 		return cc.makeHandler(spec, encoder, decoder)
diff --git a/natsrpc/doc.go b/natsrpc/doc.go
@@ -0,0 +1,2 @@
+// Package natsrpc is an rpc implemenation using nats as transport.
+package natsrpc
diff --git a/natsrpc/errors.go b/natsrpc/errors.go
@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	// ErrNCMaxReconnect is returned if nc has MaxReconnects >= 0.
+	// ErrNCMaxReconnect is returned if nats.Conn has MaxReconnects >= 0.
 	ErrNCMaxReconnect = errors.New("natsrpc.ServerConn nats.Conn should have MaxReconnects < 0")
 
 	// ErrClosed is returned when ServerConn is closed
diff --git a/natsrpc/options.go b/natsrpc/options.go
@@ -25,6 +25,7 @@ var (
 	subjectPrefixRegexp = regexp.MustCompile(`^[a-zA-Z0-9_]+$`)
 )
 
+// SCOptLogger sets logger for ServerConn.
 func SCOptLogger(logger logr.Logger) ServerConnOption {
 	return func(sc *ServerConn) error {
 		if logger == nil {
@@ -48,6 +49,7 @@ func SCOptRunner(runner taskrunner.TaskRunner) ServerConnOption {
 	}
 }
 
+// SCOptSubjectPrefix sets subject prefix in nats namespace.
 func SCOptSubjectPrefix(subjectPrefix string) ServerConnOption {
 	return func(sc *ServerConn) error {
 		if !subjectPrefixRegexp.MatchString(subjectPrefix) {
@@ -58,6 +60,7 @@ func SCOptSubjectPrefix(subjectPrefix string) ServerConnOption {
 	}
 }
 
+// SCOptGroup sets nats subscription group name.
 func SCOptGroup(group string) ServerConnOption {
 	return func(sc *ServerConn) error {
 		if group == "" {
@@ -68,6 +71,7 @@ func SCOptGroup(group string) ServerConnOption {
 	}
 }
 
+// SCOptContext sets base context used in handlers.
 func SCOptContext(ctx context.Context) ServerConnOption {
 	return func(sc *ServerConn) error {
 		if ctx == nil {
@@ -78,6 +82,7 @@ func SCOptContext(ctx context.Context) ServerConnOption {
 	}
 }
 
+// CCOptSubjectPrefix sets subject prefix in nats namespace.
 func CCOptSubjectPrefix(subjectPrefix string) ClientConnOption {
 	return func(cc *ClientConn) error {
 		if !subjectPrefixRegexp.MatchString(subjectPrefix) {
@@ -88,6 +93,7 @@ func CCOptSubjectPrefix(subjectPrefix string) ClientConnOption {
 	}
 }
 
+// CCOptTimeout sets the default timeout when invoking rpc if there is no timeout set in context.
 func CCOptTimeout(t time.Duration) ClientConnOption {
 	return func(cc *ClientConn) error {
 		if t <= 0 {
diff --git a/natsrpc/pj.go b/natsrpc/pj.go
@@ -7,6 +7,9 @@ import (
 	. "github.com/huangjunwen/nproto/v2/rpc"
 )
 
+// PbJsonClient creates an rpc client using protobuf or json for encoding/decoding:
+//   - If both input and output are proto.Message, then use protobuf.
+//   - Otherwise use json.
 func PbJsonClient(cc *ClientConn) RPCClientFunc {
 
 	pbClient := cc.Client(pjenc.DefaultPbEncoder, pjenc.DefaultPjDecoder)
@@ -24,6 +27,7 @@ func PbJsonClient(cc *ClientConn) RPCClientFunc {
 
 }
 
+// PbJsonClient creates an rpc server using protobuf or json for decoding/encoding.
 func PbJsonServer(sc *ServerConn) RPCServerFunc {
 	return sc.Server(pjenc.DefaultPjDecoder, pjenc.DefaultPjEncoder)
 }
diff --git a/natsrpc/server.go b/natsrpc/server.go
@@ -19,6 +19,7 @@ import (
 	. "github.com/huangjunwen/nproto/v2/rpc"
 )
 
+// ServerConn wraps nats.Conn into 'server side rpc connection'.
 type ServerConn struct {
 	// Immutable fields.
 	nc            *nats.Conn
@@ -36,6 +37,7 @@ type ServerConn struct {
 	methodMaps map[string]*methodMap         // svcName -> *methodMap
 }
 
+// ServerConnOption is option in creating ClientConn.
 type ServerConnOption func(*ServerConn) error
 
 // methodMap stores method info for a service.
@@ -52,6 +54,8 @@ type methodInfo struct {
 	encoder npenc.Encoder
 }
 
+// NewServerConn creates a new ServerConn. `nc` must have MaxReconnect < 0
+// (e.g. never give up trying to reconnect).
 func NewServerConn(nc *nats.Conn, opts ...ServerConnOption) (sc *ServerConn, err error) {
 
 	if nc.Opts.MaxReconnect >= 0 {
@@ -84,12 +88,14 @@ func NewServerConn(nc *nats.Conn, opts ...ServerConnOption) (sc *ServerConn, err
 	return serverConn, nil
 }
 
+// Server creates an rpc server using specified decoder and encoder.
 func (sc *ServerConn) Server(decoder npenc.Decoder, encoder npenc.Encoder) RPCServerFunc {
 	return func(spec RPCSpec, handler RPCHandler) error {
 		return sc.registHandler(spec, handler, decoder, encoder)
 	}
 }
 
+// Close shutdowns the ServerConn: closes handler runner and unsubscribe from nats.
 func (sc *ServerConn) Close() error {
 
 	sc.mu.Lock()
diff --git a/rpc/common.go b/rpc/common.go
@@ -1,4 +1,3 @@
-// Package rpc contains high level types/interfaces for rpc implementations.
 package rpc
 
 import (
diff --git a/rpc/doc.go b/rpc/doc.go
@@ -0,0 +1,2 @@
+// Package rpc contains high level types/interfaces for rpc implementations.
+package rpc
diff --git a/stanmsg/doc.go b/stanmsg/doc.go
@@ -0,0 +1,2 @@
+// Package stanmsg contains a msg implemenation using nats-streaming-server as transport.
+package stanmsg
diff --git a/stanmsg/durconn.go b/stanmsg/durconn.go
@@ -21,6 +21,7 @@ import (
 	nppbmsg "github.com/huangjunwen/nproto/v2/pb/msg"
 )
 
+// DurConn is 'durable connection' to nats-streaming-server which handles reconnect and resubscription automatically.
 type DurConn struct {
 	// Immutable fields.
 	nc                  *nats.Conn
@@ -57,10 +58,15 @@ type subscription struct {
 	decoder     npenc.Decoder
 }
 
+// DurConnOption is option in creating DurConn.
 type DurConnOption func(*DurConn) error
 
+// SubOption is option in subscription.
 type SubOption func(*subscription) error
 
+// NewDurConn creates a new DurConn. `nc` must have MaxReconnect < 0
+// (e.g. never give up trying to reconnect).
+// `clusterID` is the nats-streaming-server's cluster id.
 func NewDurConn(nc *nats.Conn, clusterID string, opts ...DurConnOption) (durConn *DurConn, err error) {
 	if nc.Opts.MaxReconnect >= 0 {
 		return nil, ErrNCMaxReconnect
@@ -100,12 +106,14 @@ func NewDurConn(nc *nats.Conn, clusterID string, opts ...DurConnOption) (durConn
 	return dc, nil
 }
 
+// Publisher creates a publisher using specified encoder.
 func (dc *DurConn) Publisher(encoder npenc.Encoder) MsgAsyncPublisherFunc {
 	return func(ctx context.Context, spec MsgSpec, msg interface{}, cb func(error)) error {
 		return dc.publishAsync(ctx, spec, msg, encoder, cb)
 	}
 }
 
+// Subscriber creates a subscriber using specified decoder.
 func (dc *DurConn) Subscriber(decoder npenc.Decoder) MsgSubscriberFunc {
 	return func(spec MsgSpec, queue string, handler MsgHandler, opts ...interface{}) error {
 		sub := &subscription{
@@ -412,6 +420,7 @@ func (dc *DurConn) msgHandler(sub *subscription) stan.MsgHandler {
 
 }
 
+// Close shutdowns the DurConn: closes handler runner and disconnects from nats-streaming-server.
 func (dc *DurConn) Close() {
 	dc.mu.Lock()
 	if dc.closed {
diff --git a/stanmsg/options.go b/stanmsg/options.go
diff --git a/stanmsg/pj.go b/stanmsg/pj.go

Original file line number	Diff line number	Diff line change
`@@ -8,13 +8,17 @@ import (`
`8`	`8`	`)`
`9`	`9`
`10`	`10`	`var (`
	`11`	`+ // DefaultLockName is the default value of PipeOptLockName.`
`11`	`12`	`DefaultLockName = "nproto.binlogmsg"`
`12`	`13`
	`14`	`+ // DefaultMaxInflight is the default value of PipeOptMaxInflight.`
`13`	`15`	`DefaultMaxInflight = 4096`
`14`	`16`
	`17`	`+ // DefaultRetryWait is the default value of PipeOptRetryWait.`
`15`	`18`	`DefaultRetryWait = 5 * time.Second`
`16`	`19`	`)`
`17`	`20`
	`21`	`+// PipeOptLogger sets logger for MsgPipe.`
`18`	`22`	`func PipeOptLogger(logger logr.Logger) MsgPipeOption {`
`19`	`23`	`return func(pipe *MsgPipe) error {`
`20`	`24`	`if logger == nil {`
`@@ -25,6 +29,8 @@ func PipeOptLogger(logger logr.Logger) MsgPipeOption {`
`25`	`29`	`}`
`26`	`30`	`}`
`27`	`31`
	`32`	`+// PipeOptLockName sets the lock name using in MySQL get lock ("SELECT GET_LOCK"):`
	`33`	`+// only one instance of pipes can run for the same lock name.`
`28`	`34`	`func PipeOptLockName(lockName string) MsgPipeOption {`
`29`	`35`	`return func(pipe *MsgPipe) error {`
`30`	`36`	`if lockName == "" {`
`@@ -35,6 +41,7 @@ func PipeOptLockName(lockName string) MsgPipeOption {`
`35`	`41`	`}`
`36`	`42`	`}`
`37`	`43`
	`44`	`+// PipeOptMaxInflight sets the max number of messages inflight (publishing).`
`38`	`45`	`func PipeOptMaxInflight(maxInflight int) MsgPipeOption {`
`39`	`46`	`return func(pipe *MsgPipe) error {`
`40`	`47`	`if maxInflight < 1 {`
`@@ -45,6 +52,7 @@ func PipeOptMaxInflight(maxInflight int) MsgPipeOption {`
`45`	`52`	`}`
`46`	`53`	`}`
`47`	`54`
	`55`	`+// PipeOptRetryWait sets the interval between retries due to all kinds of errors.`
`48`	`56`	`func PipeOptRetryWait(t time.Duration) MsgPipeOption {`
`49`	`57`	`return func(pipe *MsgPipe) error {`
`50`	`58`	`if t <= 0 {`
Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,9 @@ import (`
`10`	`10`	`. "github.com/huangjunwen/nproto/v2/msg"`
`11`	`11`	`)`
`12`	`12`
	`13`	`+// PbJsonPublisher creates a msg publisher using protobuf or json for encoding:`
	`14`	`+// - If msg is proto.Message, then use protobuf.`
	`15`	`+// - Otherwise use json.`
`13`	`16`	`func PbJsonPublisher(q sqlh.Queryer, schema, table string) MsgPublisherFunc {`
`14`	`17`
`15`	`18`	`pbPublisher := NewMsgPublisher(pjenc.DefaultPbEncoder, q, schema, table)`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+// Package msg contains high level types/interfaces for msg implementations.`
	`2`	`+package msg`
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ import (`
`4`	`4`	`"context"`
`5`	`5`	`)`
`6`	`6`
`7`		`-// MsgPublisher is used to publish messages reliably, e.g. at least once delivery.`
	`7`	`+// MsgPublisher is used to publish messages.`
`8`	`8`	`type MsgPublisher interface {`
`9`	`9`	`// Publish publishes a message to the given subject. It returns nil if success.`
`10`	`10`	`Publish(ctx context.Context, spec MsgSpec, msg interface{}) error`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+// Package natsrpc is an rpc implemenation using nats as transport.`
	`2`	`+package natsrpc`