warehouse

Documentation

Index

• Constants
• func CheckPGHealth() bool
• func GetPrevScheduledTime(syncFrequency, syncStartAt string, currTime time.Time) time.Time
• func ScheduledTimes(syncFrequency, syncStartAt string) (times []int)
• func Start()
• type ErrorResponseT
• type HandleT
  • func (wh *HandleT) Disable()
  • func (wh *HandleT) Enable()
  • func (wh *HandleT) Setup(whType string)
  • func (wh *HandleT) SyncLoadFilesToWarehouse(job *ProcessStagingFilesJobT) (err error)
• type LoadFileJobT
• type PayloadT
• type ProcessStagingFilesJobT
• type StagingFileT
• type WarehouseManager
  • func NewWhManager(destType string) (WarehouseManager, error)

Constants

const (
	MasterMode      = "master"
	SlaveMode       = "slave"
	MasterSlaveMode = "master_and_slave"
	EmbeddedMode    = "embedded"
)

warehouses worker modes

Functions

func CheckPGHealth

func CheckPGHealth() bool

func GetPrevScheduledTime

func GetPrevScheduledTime(syncFrequency, syncStartAt string, currTime time.Time) time.Time

GetPrevScheduledTime returns closest previous scheduled time eg. Syncing every 3hrs starting at 13:00 (scheduled times: 13:00, 16:00, 19:00, 22:00, 01:00, 04:00, 07:00, 10:00) prev scheduled time for current time (eg. 18:00 -> 16:00 same day, 00:30 -> 22:00 prev day)

func ScheduledTimes

func ScheduledTimes(syncFrequency, syncStartAt string) (times []int)

ScheduledTimes returns all possible start times as per schedule eg. Syncing every 3hrs starting at 13:00 (scheduled times: 13:00, 16:00, 19:00, 22:00, 01:00, 04:00, 07:00, 10:00)

func Start

func Start()

Types

type ErrorResponseT

type ErrorResponseT struct {
	Error string
}

type HandleT

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

func (*HandleT) Disable

func (wh *HandleT) Disable()

Disable disables a router:)

func (*HandleT) Enable

func (wh *HandleT) Enable()

Enable enables a router :)

func (*HandleT) Setup

func (wh *HandleT) Setup(whType string)

func (*HandleT) SyncLoadFilesToWarehouse

func (wh *HandleT) SyncLoadFilesToWarehouse(job *ProcessStagingFilesJobT) (err error)

type LoadFileJobT

type LoadFileJobT struct {
	Upload                     warehouseutils.UploadT
	StagingFile                *StagingFileT
	Schema                     map[string]map[string]string
	Warehouse                  warehouseutils.WarehouseT
	Wg                         *misc.WaitGroup
	LoadFileIDsChan            chan []int64
	TableToBucketFolderMap     map[string]string
	TableToBucketFolderMapLock *sync.RWMutex
}

type PayloadT

type PayloadT struct {
	BatchID             string
	UploadID            int64
	StagingFileID       int64
	StagingFileLocation string
	Schema              map[string]map[string]string
	SourceID            string
	DestinationID       string
	DestinationType     string
	DestinationConfig   interface{}
	LoadFileIDs         []int64
}

type ProcessStagingFilesJobT

type ProcessStagingFilesJobT struct {
	Upload    warehouseutils.UploadT
	List      []*StagingFileT
	Warehouse warehouseutils.WarehouseT
}

type StagingFileT

type StagingFileT struct {
	ID           int64
	Location     string
	SourceID     string
	Schema       json.RawMessage
	Status       string // enum
	CreatedAt    time.Time
	FirstEventAt time.Time
	LastEventAt  time.Time
}

type WarehouseManager

type WarehouseManager interface {
	Process(config warehouseutils.ConfigT) error
	CrashRecover(config warehouseutils.ConfigT) (err error)
}

func NewWhManager

func NewWhManager(destType string) (WarehouseManager, error)