databus

Documentation

Overview

Package databus is a generated protocol buffer package.

It is generated from these files:

library/queue/databus/databus.proto

It has these top-level messages:

Header
MessagePB

Index

• Variables
• type Config
• type Databus
  • func New(c *Config) *Databus
  • func (d *Databus) Close() (err error)
  • func (d *Databus) Messages() <-chan *Message
  • func (d *Databus) Send(c context.Context, k string, v interface{}) (err error)
• type Header
  • func (*Header) Descriptor() ([]byte, []int)
  • func (m *Header) GetMetadata() map[string]string
  • func (m *Header) Marshal() (dAtA []byte, err error)
  • func (m *Header) MarshalTo(dAtA []byte) (int, error)
  • func (*Header) ProtoMessage()
  • func (m *Header) Reset()
  • func (m *Header) Size() (n int)
  • func (m *Header) String() string
  • func (m *Header) Unmarshal(dAtA []byte) error
• type Message
  • func (m *Message) Commit() (err error)
• type MessagePB
  • func (*MessagePB) Descriptor() ([]byte, []int)
  • func (m *MessagePB) GetKey() string
  • func (m *MessagePB) GetMetadata() map[string]string
  • func (m *MessagePB) GetOffset() int64
  • func (m *MessagePB) GetPartition() int32
  • func (m *MessagePB) GetTimestamp() int64
  • func (m *MessagePB) GetTopic() string
  • func (m *MessagePB) GetValue() encoding_json.RawMessage
  • func (m *MessagePB) Marshal() (dAtA []byte, err error)
  • func (m *MessagePB) MarshalTo(dAtA []byte) (int, error)
  • func (*MessagePB) ProtoMessage()
  • func (m *MessagePB) Reset()
  • func (m *MessagePB) Size() (n int)
  • func (m *MessagePB) String() string
  • func (m *MessagePB) Unmarshal(dAtA []byte) error

Variables

var (
	// ErrAction action error.
	ErrAction = errors.New("action unknown")
	// ErrFull chan full
	ErrFull = errors.New("chan full")
	// ErrNoInstance no instances
	ErrNoInstance = errors.New("no databus instances found")
)

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

Types

type Config

type Config struct {
	Key          string
	Secret       string
	Group        string
	Topic        string
	Action       string // shoule be "pub" or "sub" or "pubsub"
	Buffer       int
	Name         string // redis name, for trace
	Proto        string
	Addr         string
	Auth         string
	Active       int // pool
	Idle         int // pool
	DialTimeout  xtime.Duration
	ReadTimeout  xtime.Duration
	WriteTimeout xtime.Duration
	IdleTimeout  xtime.Duration
	Direct       bool
}

Config databus config.

type Databus

type Databus struct {
	// contains filtered or unexported fields
}

Databus databus struct.

func New

func New(c *Config) *Databus

New new a databus.

func (*Databus) Close

func (d *Databus) Close() (err error)

Close close databus conn.

func (*Databus) Messages

func (d *Databus) Messages() <-chan *Message

Messages get message chan.

func (*Databus) Send

func (d *Databus) Send(c context.Context, k string, v interface{}) (err error)

Send send message to databus.

type Header

type Header struct {
	Metadata map[string]string `` /* 140-byte string literal not displayed */
}

func (*Header) Descriptor

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

func (*Header) GetMetadata

func (m *Header) GetMetadata() map[string]string

func (*Header) Marshal

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

func (*Header) MarshalTo

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

func (*Header) ProtoMessage

func (*Header) ProtoMessage()

func (*Header) Reset

func (m *Header) Reset()

func (*Header) Size

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

func (*Header) String

func (m *Header) String() string

func (*Header) Unmarshal

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

type Message

type Message struct {
	Key       string          `json:"key"`
	Value     json.RawMessage `json:"value"`
	Topic     string          `json:"topic"`
	Partition int32           `json:"partition"`
	Offset    int64           `json:"offset"`
	Timestamp int64           `json:"timestamp"`
	// contains filtered or unexported fields
}

Message Data.

func (*Message) Commit

func (m *Message) Commit() (err error)

Commit ack message.

type MessagePB

type MessagePB struct {
	Key       string                   `protobuf:"bytes,1,opt,name=key,proto3" json:"key"`
	Value     encoding_json.RawMessage `protobuf:"bytes,2,opt,name=value,proto3,casttype=encoding/json.RawMessage" json:"value"`
	Topic     string                   `protobuf:"bytes,3,opt,name=topic,proto3" json:"topic"`
	Partition int32                    `protobuf:"varint,4,opt,name=partition,proto3" json:"partition"`
	Offset    int64                    `protobuf:"varint,5,opt,name=offset,proto3" json:"offset"`
	Timestamp int64                    `protobuf:"varint,6,opt,name=timestamp,proto3" json:"timestamp"`
	Metadata  map[string]string        `` /* 140-byte string literal not displayed */
}

func (*MessagePB) Descriptor

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

func (*MessagePB) GetKey

func (m *MessagePB) GetKey() string

func (*MessagePB) GetMetadata

func (m *MessagePB) GetMetadata() map[string]string

func (*MessagePB) GetOffset

func (m *MessagePB) GetOffset() int64

func (*MessagePB) GetPartition

func (m *MessagePB) GetPartition() int32

func (*MessagePB) GetTimestamp

func (m *MessagePB) GetTimestamp() int64

func (*MessagePB) GetTopic

func (m *MessagePB) GetTopic() string

func (*MessagePB) GetValue

func (m *MessagePB) GetValue() encoding_json.RawMessage

func (*MessagePB) Marshal

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

func (*MessagePB) MarshalTo

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

func (*MessagePB) ProtoMessage

func (*MessagePB) ProtoMessage()

func (*MessagePB) Reset

func (m *MessagePB) Reset()

func (*MessagePB) Size

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

func (*MessagePB) String

func (m *MessagePB) String() string

func (*MessagePB) Unmarshal

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