http

Documentation

Index

• Constants
• func GetLookUpSource() *lookupSource
• type AccessTokenConf
• type ClientConf
  • func (cc *ClientConf) InitConf(device string, props map[string]interface{}, ...) error
• type ClientConfOption
• type PullSource
  • func (hps *PullSource) Close(ctx api.StreamContext) error
  • func (hps *PullSource) Configure(device string, props map[string]interface{}) error
  • func (hps *PullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error)
• type PushConf
• type PushSource
  • func (hps *PushSource) Close(ctx api.StreamContext) error
  • func (hps *PushSource) Configure(endpoint string, props map[string]interface{}) error
  • func (hps *PushSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error)
• type RawConf
• type RefreshTokenConf
• type RestSink
  • func (ms *RestSink) Close(ctx api.StreamContext) error
  • func (ms *RestSink) Collect(ctx api.StreamContext, item interface{}) error
  • func (ms *RestSink) CollectResend(ctx api.StreamContext, item interface{}) error
  • func (ms *RestSink) Configure(ps map[string]interface{}) error
  • func (ms *RestSink) Open(ctx api.StreamContext) error
  • func (ms *RestSink) Validate(props map[string]interface{}) error
• type WithClientConfOption
  • func WithCheckInterval(checkInterval bool) WithClientConfOption

Constants

const (
	DefaultInterval = 10000
	DefaultTimeout  = 5000
)

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

Functions

func GetLookUpSource

func GetLookUpSource() *lookupSource

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{}, withOptions ...WithClientConfOption) error

type ClientConfOption

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

type PullSource

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

func (*PullSource) Close

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

func (*PullSource) Configure

func (hps *PullSource) Configure(device string, props map[string]interface{}) error

func (*PullSource) Open

func (hps *PullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error)

type PushConf

type PushConf struct {
	Method       string `json:"method"`
	ContentType  string `json:"contentType"`
	BufferLength int    `json:"bufferLength"`
	Endpoint     string `json:"endpoint"`
}

type PushSource

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

func (*PushSource) Close

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

func (*PushSource) Configure

func (hps *PushSource) Configure(endpoint string, props map[string]interface{}) error

func (*PushSource) Open

func (hps *PushSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error)

type RawConf

type RawConf struct {
	Url       string      `json:"url"`
	Method    string      `json:"method"`
	Body      string      `json:"body"`
	BodyType  string      `json:"bodyType"`
	Headers   interface{} `json:"headers"`
	Timeout   int         `json:"timeout"`
	DebugResp bool        `json:"debugResp"`
	// Could be code or body
	ResponseType string                            `json:"responseType"`
	OAuth        map[string]map[string]interface{} `json:"oauth"`
	// source specific properties
	Interval    int    `json:"interval"`
	Incremental bool   `json:"incremental"`
	ResendUrl   string `json:"resendDestination"`
	// sink specific properties
	SendSingle bool `json:"sendSingle"`
	// inferred properties
	HeadersTemplate string
	HeadersMap      map[string]string
	Compression     string `json:"compression"` // Compression specifies the algorithms used to payload compression
}

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 (ms *RestSink) Close(ctx api.StreamContext) error

func (*RestSink) Collect

func (ms *RestSink) Collect(ctx api.StreamContext, item interface{}) error

func (*RestSink) CollectResend

func (ms *RestSink) CollectResend(ctx api.StreamContext, item interface{}) error

func (*RestSink) Configure

func (ms *RestSink) Configure(ps map[string]interface{}) error

func (*RestSink) Open

func (ms *RestSink) Open(ctx api.StreamContext) error

func (*RestSink) Validate

func (ms *RestSink) Validate(props map[string]interface{}) error

type WithClientConfOption

type WithClientConfOption func(clientConf *ClientConfOption)

func WithCheckInterval

func WithCheckInterval(checkInterval bool) WithClientConfOption