http

Documentation

Index

• Constants
• func GetSink() api.Sink
• func GetSource() api.Source
• type AccessTokenConf
• type ClientConf
  • func (cc *ClientConf) InitConf(device string, props map[string]interface{}) error
• type HttpPullSource
  • func (hps *HttpPullSource) Close(ctx api.StreamContext) error
  • func (hps *HttpPullSource) Connect(ctx api.StreamContext) error
  • func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]any) error
  • func (hps *HttpPullSource) Pull(ctx api.StreamContext, trigger time.Time, ingest api.TupleIngest, ...)
• type HttpPushSource
  • func (h *HttpPushSource) Close(ctx api.StreamContext) error
  • func (h *HttpPushSource) Connect(ctx api.StreamContext) error
  • func (h *HttpPushSource) Provision(ctx api.StreamContext, configs map[string]any) error
  • func (h *HttpPushSource) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error
• type PushConf
• type RawConf
• type RefreshTokenConf
• type RestSink
  • func (r *RestSink) Close(ctx api.StreamContext) error
  • func (r *RestSink) Collect(ctx api.StreamContext, item api.MessageTuple) error
  • func (r *RestSink) CollectList(ctx api.StreamContext, items api.MessageTupleList) error
  • func (r *RestSink) Connect(ctx api.StreamContext) error
  • func (r *RestSink) Provision(ctx api.StreamContext, configs map[string]any) error

Constants

const (
	BODY_ERR = "response body error"
	CODE_ERR = "response code error"
)

const (
	DefaultTimeout = 5000 * time.Millisecond
)

Functions

func GetSink

func GetSink() api.Sink

func GetSource

func GetSource() api.Source

Types

type AccessTokenConf

type AccessTokenConf struct {
	Url            string `json:"url"`
	Body           string `json:"body"`
	Expire         string `json:"expire"`
	ExpireInSecond int
}

type ClientConf

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

ClientConf is the configuration for http client It is shared by httppull source and rest sink to configure their http client

func (*ClientConf) InitConf

func (cc *ClientConf) InitConf(device string, props map[string]interface{}) error

type HttpPullSource

type HttpPullSource struct {
	*ClientConf
	// contains filtered or unexported fields
}

func (*HttpPullSource) Close

func (hps *HttpPullSource) Close(ctx api.StreamContext) error

func (*HttpPullSource) Connect

func (hps *HttpPullSource) Connect(ctx api.StreamContext) error

func (*HttpPullSource) Provision

func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]any) error

func (*HttpPullSource) Pull

func (hps *HttpPullSource) Pull(ctx api.StreamContext, trigger time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest)

type HttpPushSource

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

func (*HttpPushSource) Close

func (h *HttpPushSource) Close(ctx api.StreamContext) error

func (*HttpPushSource) Connect

func (h *HttpPushSource) Connect(ctx api.StreamContext) error

func (*HttpPushSource) Provision

func (h *HttpPushSource) Provision(ctx api.StreamContext, configs map[string]any) error

func (*HttpPushSource) Subscribe

func (h *HttpPushSource) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error

type PushConf

type PushConf struct {
	Method       string `json:"method"`
	BufferLength int    `json:"bufferLength"`
	DataSource   string `json:"datasource"`
}

type RawConf

type RawConf struct {
	Url         string            `json:"url"`
	Method      string            `json:"method"`
	Body        string            `json:"body"`
	BodyType    string            `json:"bodyType"`
	Headers     map[string]string `json:"headers"`
	Timeout     cast.DurationConf `json:"timeout"`
	Incremental bool              `json:"incremental"`

	OAuth map[string]map[string]interface{} `json:"oauth"`
	// Could be code or body
	ResponseType string `json:"responseType"`
	Compression  string `json:"compression"` // Compression specifies the algorithms used to payload compression

	DebugResp bool `json:"debugResp"`
}

type RefreshTokenConf

type RefreshTokenConf struct {
	Url     string            `json:"url"`
	Headers map[string]string `json:"headers"`
	Body    string            `json:"body"`
}

type RestSink

type RestSink struct {
	*ClientConf
}

func (*RestSink) Close

func (r *RestSink) Close(ctx api.StreamContext) error

func (*RestSink) Collect

func (r *RestSink) Collect(ctx api.StreamContext, item api.MessageTuple) error

func (*RestSink) CollectList

func (r *RestSink) CollectList(ctx api.StreamContext, items api.MessageTupleList) error

func (*RestSink) Connect

func (r *RestSink) Connect(ctx api.StreamContext) error

func (*RestSink) Provision

func (r *RestSink) Provision(ctx api.StreamContext, configs map[string]any) error