workflow

Documentation

Overview

Package workflow contains an experimental workflow for Archivemica transfers.

It's not generalized since it contains client-specific activities. However, the long-term goal is to build a system where workflows and activities are dynamically set up based on user input.

Index

• Variables
• type AsyncCompletionActivity
  • func NewAsyncCompletionActivity(m *manager.Manager) *AsyncCompletionActivity
  • func (a *AsyncCompletionActivity) Execute(ctx context.Context, colID uint) (string, error)
• type ProcessingWorkflow
  • func NewProcessingWorkflow(m *manager.Manager) *ProcessingWorkflow
  • func (w *ProcessingWorkflow) Execute(ctx workflow.Context, req *collection.ProcessingWorkflowRequest) error
  • func (w *ProcessingWorkflow) SessionHandler(sessCtx workflow.Context, attempt int, tinfo *TransferInfo, ...) error
• type TransferInfo

Variables

var AsyncCompletionActivityName = "async-completion-activity"

var ErrAsyncCompletionAbandoned = errors.New("user abandoned")

Types

type AsyncCompletionActivity

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

func NewAsyncCompletionActivity

func NewAsyncCompletionActivity(m *manager.Manager) *AsyncCompletionActivity

func (*AsyncCompletionActivity) Execute

func (a *AsyncCompletionActivity) Execute(ctx context.Context, colID uint) (string, error)

type ProcessingWorkflow

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

func NewProcessingWorkflow

func NewProcessingWorkflow(m *manager.Manager) *ProcessingWorkflow

func (*ProcessingWorkflow) Execute

func (w *ProcessingWorkflow) Execute(ctx workflow.Context, req *collection.ProcessingWorkflowRequest) error

ProcessingWorkflow orchestrates all the activities related to the processing of a SIP in Archivematica, including is retrieval, creation of transfer, etc...

Retrying this workflow would result in a new Archivematica transfer. We do not have a retry policy in place. The user could trigger a new instance via the API.

func (*ProcessingWorkflow) SessionHandler

func (w *ProcessingWorkflow) SessionHandler(sessCtx workflow.Context, attempt int, tinfo *TransferInfo, nameInfo nha.NameInfo, validationConfig validation.Config) error

SessionHandler runs activities that belong to the same session.

type TransferInfo

type TransferInfo struct {

	// TempFile is the temporary location where the blob is downloaded.
	//
	// It is populated by the workflow with the result of DownloadActivity.
	TempFile string

	// TransferID given by Archivematica.
	//
	// It is populated by TransferActivity.
	TransferID string

	// SIPID given by Archivematica.
	//
	// It is populated by PollTransferActivity.
	SIPID string

	// Enduro internal collection ID.
	// The zero value represents a new collection. It can be used to indicate
	// an existing collection in retries.
	//
	// It is populated via the workflow request or createPackageLocalActivity.
	CollectionID uint

	// Name of the watcher that received this blob.
	//
	// It is populated via the workflow request.
	WatcherName string

	// Pipeline name.
	//
	// It is populated via the workflow request.
	PipelineName string

	// Retention period.
	// Period of time to schedule the deletion of the original blob from the
	// watched data source. nil means no deletion.
	//
	// It is populated via the workflow request.
	RetentionPeriod *time.Duration

	// Whether the top-level directory is meant to be stripped.
	//
	// It is populated via the workflow request.
	StripTopLevelDir bool

	// Key of the blob.
	//
	// It is populated via the workflow request.
	Key string

	// Batch directory that contains the blob.
	//
	// It is populated via the workflow request.
	BatchDir string

	// StoredAt is the time when the AIP is stored.
	//
	// It is populated by PollIngestActivity as long as Ingest completes.
	StoredAt time.Time

	// PipelineConfig is the configuration of the pipeline that this workflow
	// uses to provide access to its activities.
	//
	// It is populated by loadConfigLocalActivity.
	PipelineConfig *pipeline.Config

	// PipelineID is the UUID of the Archivematica pipeline. Extracted from
	// the API response header when the transfer is submitted.
	//
	// It is populated by transferActivity.
	PipelineID string

	// Hooks is the hook config store.
	//
	// It is populated by loadConfigLocalActivity.
	Hooks map[string]map[string]interface{}

	// Information about the bundle (transfer) that we submit to Archivematica.
	// Full path, relative path...
	//
	// It is populated by BundleActivity.
	Bundle activities.BundleActivityResult
}

TransferInfo is shared state that is passed down to activities. It can be useful for hooks that may require quick access to processing state. TODO: clean this up, e.g.: it can embed a collection.Collection.