pipeline

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(
	ctx context.Context,
	logger log.Logger,
	cfg *service.Config,
	shardRepository shards.Repository,
	fileRepository files.Repository,
	httpFiles stream.Subscription,
) (*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 ¶

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

	Err error
	// contains filtered or unexported fields
}

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, 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(logger log.Logger, shard service.Shard, cfg service.UploadAgents) (XferMerging, error)