lib

Documentation

Index

• func CreateStateJSON()
• func GenerateRecordMessage(record interface{}) error
• func GenerateSchema(records []interface{}) (map[string]interface{}, error)
• func GenerateSchemaMessage(schema map[string]interface{}) error
• func GenerateStateMessage(state *State) error
• func ParseRecord(record []byte, resultChan chan<- *interface{})
• func UpdateState(record interface{})
• type Config
• type Message
• type State
  • func ParseStateJSON() (*State, error)

Functions

func CreateStateJSON

func CreateStateJSON()

/////////////////////////////////////////////////////////// CREATE state_<STREAM>.json ///////////////////////////////////////////////////////////

func GenerateRecordMessage

func GenerateRecordMessage(record interface{}) error

func GenerateSchema

func GenerateSchema(records []interface{}) (map[string]interface{}, error)

/////////////////////////////////////////////////////////// GENERATE SCHEMA Infer the schema from records ///////////////////////////////////////////////////////////

func GenerateSchemaMessage

func GenerateSchemaMessage(schema map[string]interface{}) error

func GenerateStateMessage

func GenerateStateMessage(state *State) error

func ParseRecord

func ParseRecord(record []byte, resultChan chan<- *interface{})

///////////////////////////////////////////////////////// PARSE RECORD processRecord() and send to resultChan /////////////////////////////////////////////////////////

func UpdateState

func UpdateState(record interface{})

/////////////////////////////////////////////////////////// UPDATE state_<STREAM>.json ///////////////////////////////////////////////////////////

Types

type Config

type Config struct {
	StreamName     *string `json:"stream_name,omitempty"`
	SourceType     *string `json:"source_type,omitempty"`
	URL            *string `json:"url,omitempty"`
	MaxConcurrency *int    `json:"max_concurrency,omitempty"`
	Records        *struct {
		UniqueKeyPath       *[]string   `json:"unique_key_path,omitempty"`
		DropFieldPaths      *[][]string `json:"drop_field_paths,omitempty"`
		SensitiveFieldPaths *[][]string `json:"sensitive_field_paths,omitempty"`
	} `json:"records,omitempty"`
	Database *struct {
		Table *string `json:"table,omitempty"`
	} `json:"db,omitempty"`
	Rest *struct {
		Sleep *int `json:"sleep,omitempty"`
		Auth  *struct {
			Required *bool   `json:"required,omitempty"`
			Strategy *string `json:"strategy,omitempty"`
			Basic    *struct {
				Username *string `json:"username,omitempty"`
				Password *string `json:"password,omitempty"`
			} `json:"basic,omitempty"`
			Token *struct {
				Header      *string `json:"header,omitempty"`
				HeaderValue *string `json:"header_value,omitempty"`
			} `json:"token,omitempty"`
			Oauth *struct {
				ClientID     *string `json:"client_id,omitempty"`
				ClientSecret *string `json:"client_secret,omitempty"`
				RefreshToken *string `json:"refresh_token,omitempty"`
				TokenURL     *string `json:"token_url,omitempty"`
			} `json:"oauth,omitempty"`
		} `json:"auth,omitempty"`
		Response *struct {
			RecordsPath        *[]string `json:"records_path,omitempty"`
			Pagination         *bool     `json:"pagination,omitempty"`
			PaginationStrategy *string   `json:"pagination_strategy,omitempty"`
			PaginationNextPath *[]string `json:"pagination_next_path,omitempty"`
			PaginationQuery    *struct {
				QueryParameter *string `json:"query_parameter,omitempty"`
				QueryValue     *int    `json:"query_value,omitempty"`
				QueryIncrement *int    `json:"query_increment,omitempty"`
			} `json:"pagination_query,omitempty"`
		} `json:"response,omitempty"`
	} `json:"rest,omitempty"`
}

///////////////////////////////////////////////////////// CONFIG.JSON Parse config.json file to Config struct /////////////////////////////////////////////////////////

var ParsedConfig Config

type Message

type Message struct {
	Type               string                 `json:"type"`
	Record             map[string]interface{} `json:"record,omitempty"`
	Stream             string                 `json:"stream,omitempty"`
	Schema             interface{}            `json:"schema,omitempty"`
	Value              interface{}            `json:"value,omitempty"`
	KeyProperties      []string               `json:"key_properties,omitempty"`
	BookmarkProperties []string               `json:"bookmark_properties,omitempty"`
}

///////////////////////////////////////////////////////// MESSAGES Generate schema, record & state messages /////////////////////////////////////////////////////////

type State

type State struct {
	Type  string `json:"type"`
	Value struct {
		Bookmarks map[string]struct {
			BookmarkUpdatedAt string   `json:"bookmark_updated_at"`
			Bookmark          []string `json:"bookmark"`
		} `json:"bookmarks"`
	} `json:"Value"`
}

///////////////////////////////////////////////////////// STATE_<STREAM>.JSON /////////////////////////////////////////////////////////

var ParsedState *State

func ParseStateJSON

func ParseStateJSON() (*State, error)

/////////////////////////////////////////////////////////// PARSE state_<STREAM>.json ///////////////////////////////////////////////////////////