http

Documentation

Index

• Constants
• func GetLookUpSource() api.Source
• func GetSink() api.Sink
• func GetSource() api.Source
• type AccessTokenConf
• type ClientConf
  • func (cc *ClientConf) InitConf(device string, props map[string]interface{}) error
• type HttpLookupSource
  • func (hls *HttpLookupSource) Close(ctx api.StreamContext) error
  • func (hls *HttpLookupSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error
  • func (hls *HttpLookupSource) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error)
  • func (hls *HttpLookupSource) Provision(ctx api.StreamContext, configs map[string]any) error
• type HttpPullSource
  • func (hps *HttpPullSource) Close(ctx api.StreamContext) error
  • func (hps *HttpPullSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) 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, sch api.StatusChangeHandler) 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.RawTuple) error
  • func (r *RestSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) 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 GetLookUpSource

func GetLookUpSource() api.Source

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"`
	Headers        map[string]string `json:"headers"`
	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 HttpLookupSource

type HttpLookupSource struct {
	*ClientConf
}

func (*HttpLookupSource) Close

func (hls *HttpLookupSource) Close(ctx api.StreamContext) error

func (*HttpLookupSource) Connect

func (hls *HttpLookupSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error

func (*HttpLookupSource) Lookup

func (hls *HttpLookupSource) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error)

func (*HttpLookupSource) Provision

func (hls *HttpLookupSource) Provision(ctx api.StreamContext, configs map[string]any) 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, sch api.StatusChangeHandler) 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, sch api.StatusChangeHandler) 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"`
	Format      string            `json:"format"`
	Headers     map[string]string `json:"headers"`
	Timeout     cast.DurationConf `json:"timeout"`
	Incremental bool              `json:"incremental"`

	OAuth      map[string]map[string]interface{} `json:"oauth"`
	SendSingle bool                              `json:"sendSingle"`
	// 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
	// contains filtered or unexported fields
}

func (*RestSink) Close

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

func (*RestSink) Collect

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

func (*RestSink) Connect

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

func (*RestSink) Provision

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