123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116 |
- package yomo
- import (
- "context"
- "github.com/yomorun/yomo/core"
- "github.com/yomorun/yomo/core/frame"
- "github.com/yomorun/yomo/pkg/id"
- )
- // Source is responsible for sending data to yomo.
- type Source interface {
- // Close will close the connection to YoMo-Zipper.
- Close() error
- // Connect to YoMo-Zipper.
- Connect() error
- // Write the data to directed downstream.
- Write(tag uint32, data []byte) error
- // WriteWithTarget writes data to sfn instance with specified target.
- WriteWithTarget(tag uint32, data []byte, target string) error
- // SetErrorHandler set the error handler function when server error occurs
- SetErrorHandler(fn func(err error))
- }
- // YoMo-Source
- type yomoSource struct {
- name string
- zipperAddr string
- client *core.Client
- }
- var _ Source = &yomoSource{}
- // NewSource create a yomo-source
- func NewSource(name, zipperAddr string, opts ...SourceOption) Source {
- clientOpts := make([]core.ClientOption, len(opts))
- for k, v := range opts {
- clientOpts[k] = core.ClientOption(v)
- }
- client := core.NewClient(name, zipperAddr, core.ClientTypeSource, clientOpts...)
- client.Logger = client.Logger.With(
- "component", core.ClientTypeSource.String(),
- "source_id", client.ClientID(),
- "source_name", client.Name(),
- "zipper_addr", zipperAddr,
- )
- return &yomoSource{
- name: name,
- zipperAddr: zipperAddr,
- client: client,
- }
- }
- // Close will close the connection to YoMo-Zipper.
- func (s *yomoSource) Close() error {
- _ = s.client.Close()
- s.client.Logger.Debug("the source is closed")
- return nil
- }
- // Connect to YoMo-Zipper.
- func (s *yomoSource) Connect() error {
- return s.client.Connect(context.Background())
- }
- // Write writes data with specified tag.
- func (s *yomoSource) Write(tag uint32, data []byte) error {
- if err := frame.IsReservedTag(tag); err != nil {
- return err
- }
- md := core.NewMetadata(s.client.ClientID(), id.New())
- mdBytes, err := md.Encode()
- // metadata
- if err != nil {
- return err
- }
- f := &frame.DataFrame{
- Tag: tag,
- Metadata: mdBytes,
- Payload: data,
- }
- s.client.Logger.Debug("source write", "tag", tag, "dataLen", len(data))
- return s.client.WriteFrame(f)
- }
- // WritePayload writes `yomo.Payload` with specified tag.
- func (s *yomoSource) WriteWithTarget(tag uint32, data []byte, target string) error {
- if err := frame.IsReservedTag(tag); err != nil {
- return err
- }
- md := core.NewMetadata(s.client.ClientID(), id.New())
- if target != "" {
- core.SetMetadataTarget(md, target)
- }
- mdBytes, err := md.Encode()
- if err != nil {
- return err
- }
- f := &frame.DataFrame{
- Tag: tag,
- Metadata: mdBytes,
- Payload: data,
- }
- s.client.Logger.Debug("source write with target", "tag", tag, "data", data, "target", target)
- return s.client.WriteFrame(f)
- }
- // SetErrorHandler set the error handler function when server error occurs
- func (s *yomoSource) SetErrorHandler(fn func(err error)) {
- s.client.SetErrorHandler(fn)
- }
|