pipeline

Documentation

Index

• type FileReceiver
  • func Start(ctx context.Context, logger log.Logger, cfg *service.Config, ...) (*FileReceiver, error)
  • func (fr *FileReceiver) RegisterAdminRoutes(r *admin.Server)
  • func (fr *FileReceiver) ReplaceStreamFiles(sub stream.Subscription)
  • func (fr *FileReceiver) Shutdown()
  • func (fr *FileReceiver) Start(ctx context.Context)
• type MockXferMerging
  • func (merge *MockXferMerging) HandleCancel(_ context.Context, cancel incoming.CancelACHFile) (incoming.FileCancellationResponse, error)
  • func (merge *MockXferMerging) HandleXfer(_ context.Context, xfer incoming.ACHFile) error
  • func (merge *MockXferMerging) WithEachMerged(ctx context.Context, ...) (mergedFiles, error)
• type XferMerging
  • func NewMerging(logger log.Logger, shard service.Shard, cfg service.UploadAgents) (XferMerging, error)

Types

type FileReceiver

type FileReceiver struct {
	CancellationResponses chan models.FileCancellationResponse
	// contains filtered or unexported fields
}

FileReceiver accepts an ACH file from a number of pubsub Subscriptions and finds the appropriate aggregator for the shardKey.

func Start

func Start(
	ctx context.Context,
	logger log.Logger,
	cfg *service.Config,
	shardRepository shards.Repository,
	fileRepository files.Repository,
	httpFiles stream.Subscription,
) (*FileReceiver, error)

func (*FileReceiver) RegisterAdminRoutes

func (fr *FileReceiver) RegisterAdminRoutes(r *admin.Server)

func (*FileReceiver) ReplaceStreamFiles

func (fr *FileReceiver) ReplaceStreamFiles(sub stream.Subscription)

func (*FileReceiver) Shutdown

func (fr *FileReceiver) Shutdown()

func (*FileReceiver) Start

func (fr *FileReceiver) Start(ctx context.Context)

type MockXferMerging

type MockXferMerging struct {
	LatestFile           *incoming.ACHFile
	LatestCancel         *incoming.CancelACHFile
	CancellationResponse incoming.FileCancellationResponse

	Err error
	// contains filtered or unexported fields
}

func (*MockXferMerging) HandleCancel

func (merge *MockXferMerging) HandleCancel(_ context.Context, cancel incoming.CancelACHFile) (incoming.FileCancellationResponse, error)

func (*MockXferMerging) HandleXfer

func (merge *MockXferMerging) HandleXfer(_ context.Context, xfer incoming.ACHFile) error

func (*MockXferMerging) WithEachMerged

func (merge *MockXferMerging) WithEachMerged(ctx context.Context, f func(context.Context, int, upload.Agent, *ach.File) (string, error)) (mergedFiles, error)

type XferMerging

type XferMerging interface {
	HandleXfer(ctx context.Context, xfer incoming.ACHFile) error
	HandleCancel(ctx context.Context, cancel incoming.CancelACHFile) (incoming.FileCancellationResponse, error)

	WithEachMerged(ctx context.Context, f func(context.Context, int, upload.Agent, *ach.File) (string, error)) (mergedFiles, error)
}

XferMerging represents logic for accepting ACH files to be merged together.

The idea is to take Xfers and store them on a filesystem (or other durable storage) prior to a cutoff window. The specific storage could be based on the FileHeader.

On the cutoff trigger WithEachMerged is called to merge files together and offer each merged file for an upload.

func NewMerging

func NewMerging(logger log.Logger, shard service.Shard, cfg service.UploadAgents) (XferMerging, error)