processing

README

Pipeline

The processing pipeline works in these stages:

• Collections: Read the config that define each collection
• Collections Manifest: For each collection, process (in-order) the files specified
• Unzip: Unzip a file for a collection
• Read: Read/parse the content of a file
• Generate: Generate meaningful records from parsed file content
• Persist: Persist valid records

Documentation

Index

• func Process(ins []<-chan *ProcessingPayload, destinationDirectory string, ...)
• type Collection
• type ImportManifest
• type ProcessingContext
• type ProcessingPayload
• type ProcessingStrategy
• type ProcessorHandler

Functions

func Process

func Process(ins []<-chan *ProcessingPayload, destinationDirectory string, deleteDestinationDirectory bool, dataStore *data_store.DataStore)

Types

type Collection

type Collection struct {
	Collection       string                 `json:"collection"`
	ComparisonFields []string               `json:"comparisonFields"`
	Mappings         map[string]interface{} `json:"mappings"`
}

type ImportManifest

type ImportManifest struct {
	SourceFiles []struct {
		Zip       string `json:"zip"`
		PathInZip string `json:"pathInZip"`
	} `json:"sourceFiles"`
}

type ProcessingContext

type ProcessingContext struct {
	Label           string
	RateTracker     *stats.RateTracker
	DataStore       *data_store.DataStore
	StatsTracker    *stats.StatsTracker
	ProcessorsTotal int
	Strategy        ProcessingStrategy
}

type ProcessingPayload

type ProcessingPayload struct {
	Value interface{}
}

type ProcessingStrategy

type ProcessingStrategy string

const (
	WaitUntilAllDone ProcessingStrategy = "WaitUntilAllDone"
	SerialPerInput   ProcessingStrategy = "SerialPerInput"
)

type ProcessorHandler

type ProcessorHandler func(ctx *ProcessingContext, in <-chan *ProcessingPayload) <-chan *ProcessingPayload