filesystem

README

Kubemq Filesystem Source

Kubemq Filesystem source sends .

Prerequisites

The following are required to run events source connector:

• kubemq cluster
• kubemq-sources deployment

Configuration

Filesystem source connector configuration properties:

Properties Key	Required	Description	Example
folders	yes	set list of folders to watch	"c:/folder1,c:/folder2"
bucket_type	yes	set remote target bucket type	"aws", "gcp", "minio", "filesystem","hdfs","azure","pass-through"
bucket_name	yes	set remote target bucket/dir name	"dir1"
concurrency	no	set sending concurrency	"1"
backup_folder	no	set backup folder for files after send	"1"
scan_interval	no	set filesystem scan interval in sec	"5"

Example:

bindings:
- name: fs
  source:
    kind: storage.filesystem
    properties:
      folders: 'd:\test\source,d:\test\folder2'
      bucket_type: aws
      bucket_name: aws_bucket_name
      concurrency: 1
      backup_folder: 'd:\backup'
      scan_interval: 5
  target:
    kind: kubemq.queue
    properties:
      address: localhost:50000
      channel: queue.fs
  properties: {}

Documentation

Index

• func Connector() *common.Connector
• type Client
  • func New() *Client
  • func (c *Client) Connector() *common.Connector
  • func (c *Client) Init(ctx context.Context, cfg config.Spec, log *logger.Logger) error
  • func (c *Client) Start(ctx context.Context, target middleware.Middleware) error
  • func (c *Client) Stop() error
• type SourceFile
  • func NewSourceFile(info os.FileInfo, path string, root string, movePath string) *SourceFile
  • func (s *SourceFile) Delete() error
  • func (s *SourceFile) Do() error
  • func (s *SourceFile) FileDir() string
  • func (s *SourceFile) FileName() string
  • func (s *SourceFile) FullPath() string
  • func (s *SourceFile) Hash() string
  • func (s *SourceFile) Load() ([]byte, error)
  • func (s *SourceFile) Metadata() string
  • func (s *SourceFile) Request(bucketType string, bucketName string) (*types.Request, error)
• type TargetsMetadata
  • func NewTargetMetadata() TargetsMetadata
  • func (m TargetsMetadata) Set(key, value string) TargetsMetadata
  • func (m TargetsMetadata) String() string
• type TargetsRequest
  • func NewTargetsRequest() *TargetsRequest
  • func ParseRequest(body []byte) (*TargetsRequest, error)
  • func (r *TargetsRequest) MarshalBinary() []byte
  • func (r *TargetsRequest) SetData(value []byte) *TargetsRequest
  • func (r *TargetsRequest) SetMetadata(value TargetsMetadata) *TargetsRequest
  • func (r *TargetsRequest) SetMetadataKeyValue(key, value string) *TargetsRequest
  • func (r *TargetsRequest) Size() float64

Functions

func Connector

func Connector() *common.Connector

Types

type Client

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

func New

func New() *Client

func (*Client) Connector

func (c *Client) Connector() *common.Connector

func (*Client) Init

func (c *Client) Init(ctx context.Context, cfg config.Spec, log *logger.Logger) error

func (*Client) Start

func (c *Client) Start(ctx context.Context, target middleware.Middleware) error

func (*Client) Stop

func (c *Client) Stop() error

type SourceFile

type SourceFile struct {
	Info     os.FileInfo
	Path     string
	Root     string
	MovePath string
}

func NewSourceFile

func NewSourceFile(info os.FileInfo, path string, root string, movePath string) *SourceFile

func (*SourceFile) Delete

func (s *SourceFile) Delete() error

func (*SourceFile) Do

func (s *SourceFile) Do() error

func (*SourceFile) FileDir

func (s *SourceFile) FileDir() string

func (*SourceFile) FileName

func (s *SourceFile) FileName() string

func (*SourceFile) FullPath

func (s *SourceFile) FullPath() string

func (*SourceFile) Hash

func (s *SourceFile) Hash() string

func (*SourceFile) Load

func (s *SourceFile) Load() ([]byte, error)

func (*SourceFile) Metadata

func (s *SourceFile) Metadata() string

func (*SourceFile) Request

func (s *SourceFile) Request(bucketType string, bucketName string) (*types.Request, error)

type TargetsMetadata

type TargetsMetadata map[string]string

func NewTargetMetadata

func NewTargetMetadata() TargetsMetadata

func (TargetsMetadata) Set

func (m TargetsMetadata) Set(key, value string) TargetsMetadata

func (TargetsMetadata) String

func (m TargetsMetadata) String() string

type TargetsRequest

type TargetsRequest struct {
	Metadata TargetsMetadata `json:"metadata,omitempty"`
	Data     []byte          `json:"data,omitempty"`
}

func NewTargetsRequest

func NewTargetsRequest() *TargetsRequest

func ParseRequest

func ParseRequest(body []byte) (*TargetsRequest, error)

func (*TargetsRequest) MarshalBinary

func (r *TargetsRequest) MarshalBinary() []byte

func (*TargetsRequest) SetData

func (r *TargetsRequest) SetData(value []byte) *TargetsRequest

func (*TargetsRequest) SetMetadata

func (r *TargetsRequest) SetMetadata(value TargetsMetadata) *TargetsRequest

func (*TargetsRequest) SetMetadataKeyValue

func (r *TargetsRequest) SetMetadataKeyValue(key, value string) *TargetsRequest

func (*TargetsRequest) Size

func (r *TargetsRequest) Size() float64