activities

Documentation

Overview

Package activities implements Enduro's workflow activities.

Index

• Constants
• type AcquirePipelineActivity
  • func NewAcquirePipelineActivity(pipelineRegistry *pipeline.Registry) *AcquirePipelineActivity
  • func (a *AcquirePipelineActivity) Execute(ctx context.Context, pipelineName string) error
• type BundleActivity
  • func NewBundleActivity(m *manager.Manager) *BundleActivity
  • func (a *BundleActivity) Bundle(ctx context.Context, unar archiver.Unarchiver, ...) (string, string, error)
  • func (a *BundleActivity) Copy(ctx context.Context, src, dst string, stripTopLevelDir bool, ...) (string, string, error)
  • func (a *BundleActivity) Execute(ctx context.Context, params *BundleActivityParams) (*BundleActivityResult, error)
  • func (a *BundleActivity) SingleFile(ctx context.Context, transferDir, key, tempFile string) (string, error)
  • func (a *BundleActivity) Unarchiver(key, filename string) archiver.Unarchiver
• type BundleActivityParams
• type BundleActivityResult
• type CleanUpActivity
  • func NewCleanUpActivity(m *manager.Manager) *CleanUpActivity
  • func (a *CleanUpActivity) Execute(ctx context.Context, params *CleanUpActivityParams) error
• type CleanUpActivityParams
• type DeleteOriginalActivity
  • func NewDeleteOriginalActivity(m *manager.Manager) *DeleteOriginalActivity
  • func (a *DeleteOriginalActivity) Execute(ctx context.Context, watcherName, batchDir, key string) error
• type DisposeOriginalActivity
  • func NewDisposeOriginalActivity(m *manager.Manager) *DisposeOriginalActivity
  • func (a *DisposeOriginalActivity) Execute(ctx context.Context, watcherName, completedDir, batchDir, key string) error
• type DownloadActivity
  • func NewDownloadActivity(m *manager.Manager) *DownloadActivity
  • func (a *DownloadActivity) Execute(ctx context.Context, pipelineName, watcherName, key string) (string, error)
• type HidePackageActivity
  • func NewHidePackageActivity(m *manager.Manager) *HidePackageActivity
  • func (a *HidePackageActivity) Execute(ctx context.Context, unitID, unitType, pipelineName string) error
• type PollIngestActivity
  • func NewPollIngestActivity(pipelineRegistry *pipeline.Registry) *PollIngestActivity
  • func (a *PollIngestActivity) Execute(ctx context.Context, params *PollIngestActivityParams) (time.Time, error)
• type PollIngestActivityParams
• type PollTransferActivity
  • func NewPollTransferActivity(pipelineRegistry *pipeline.Registry) *PollTransferActivity
  • func (a *PollTransferActivity) Execute(ctx context.Context, params *PollTransferActivityParams) (string, error)
• type PollTransferActivityParams
• type PopulateMetadataActivity
  • func NewPopulateMetadataActivity(pipelineRegistry *pipeline.Registry) *PopulateMetadataActivity
  • func (a *PopulateMetadataActivity) Execute(ctx context.Context, params *PopulateMetadataActivityParams) error
• type PopulateMetadataActivityParams
• type TransferActivity
  • func NewTransferActivity(m *manager.Manager) *TransferActivity
  • func (a *TransferActivity) Execute(ctx context.Context, params *TransferActivityParams) (*TransferActivityResponse, error)
• type TransferActivityParams
• type TransferActivityResponse
• type ValidateTransferActivity
  • func NewValidateTransferActivity() *ValidateTransferActivity
  • func (a *ValidateTransferActivity) Execute(ctx context.Context, params *ValidateTransferActivityParams) error
• type ValidateTransferActivityParams

Constants

const (
	AcquirePipelineActivityName  = "acquire-pipeline-activity"
	DownloadActivityName         = "download-activity"
	BundleActivityName           = "bundle-activity"
	TransferActivityName         = "transfer-activity"
	PollTransferActivityName     = "poll-transfer-activity"
	PollIngestActivityName       = "poll-ingest-activity"
	CleanUpActivityName          = "clean-up-activity"
	HidePackageActivityName      = "hide-package-activity"
	DeleteOriginalActivityName   = "delete-original-activity"
	DisposeOriginalActivityName  = "dispose-original-activity"
	ValidateTransferActivityName = "validate-transfer-activity"
	PopulateMetadataActivityName = "populate-metadata-activity"
)

Types

type AcquirePipelineActivity

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

AcquirePipelineActivity acquires a lock in the weighted semaphore associated to a particular pipeline.

func NewAcquirePipelineActivity

func NewAcquirePipelineActivity(pipelineRegistry *pipeline.Registry) *AcquirePipelineActivity

func (*AcquirePipelineActivity) Execute

func (a *AcquirePipelineActivity) Execute(ctx context.Context, pipelineName string) error

type BundleActivity

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

func NewBundleActivity

func NewBundleActivity(m *manager.Manager) *BundleActivity

func (*BundleActivity) Bundle

func (a *BundleActivity) Bundle(ctx context.Context, unar archiver.Unarchiver, transferDir, key, tempFile string, stripTopLevelDir bool) (string, string, error)

Bundle a transfer with the contents found in the archive.

func (*BundleActivity) Copy

func (a *BundleActivity) Copy(ctx context.Context, src, dst string, stripTopLevelDir bool, excludeHiddenFiles bool) (string, string, error)

Copy a transfer in the given destination using an intermediate temp. directory.

func (*BundleActivity) Execute

func (a *BundleActivity) Execute(ctx context.Context, params *BundleActivityParams) (*BundleActivityResult, error)

func (*BundleActivity) SingleFile

func (a *BundleActivity) SingleFile(ctx context.Context, transferDir, key, tempFile string) (string, error)

SingleFile bundles a transfer with the downloaded blob in it.

func (*BundleActivity) Unarchiver

func (a *BundleActivity) Unarchiver(key, filename string) archiver.Unarchiver

Unarchiver returns the unarchiver suited for the archival format.

type BundleActivityParams

type BundleActivityParams struct {
	WatcherName        string
	TransferDir        string
	Key                string
	TempFile           string
	StripTopLevelDir   bool
	ExcludeHiddenFiles bool
	IsDir              bool
	BatchDir           string
	Unbag              bool
}

type BundleActivityResult

type BundleActivityResult struct {
	RelPath             string // Path of the transfer relative to the transfer directory.
	FullPath            string // Full path to the transfer in the worker running the session.
	FullPathBeforeStrip string // Same as FullPath but includes the top-level dir even when stripped.
}

type CleanUpActivity

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

CleanUpActivity removes the contents that we've created in the TS location.

func NewCleanUpActivity

func NewCleanUpActivity(m *manager.Manager) *CleanUpActivity

func (*CleanUpActivity) Execute

func (a *CleanUpActivity) Execute(ctx context.Context, params *CleanUpActivityParams) error

type CleanUpActivityParams

type CleanUpActivityParams struct {
	FullPath string
}

type DeleteOriginalActivity

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

func NewDeleteOriginalActivity

func NewDeleteOriginalActivity(m *manager.Manager) *DeleteOriginalActivity

func (*DeleteOriginalActivity) Execute

func (a *DeleteOriginalActivity) Execute(ctx context.Context, watcherName, batchDir, key string) error

type DisposeOriginalActivity

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

func NewDisposeOriginalActivity

func NewDisposeOriginalActivity(m *manager.Manager) *DisposeOriginalActivity

func (*DisposeOriginalActivity) Execute

func (a *DisposeOriginalActivity) Execute(ctx context.Context, watcherName, completedDir, batchDir, key string) error

type DownloadActivity

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

DownloadActivity downloads the blob into the pipeline processing directory.

func NewDownloadActivity

func NewDownloadActivity(m *manager.Manager) *DownloadActivity

func (*DownloadActivity) Execute

func (a *DownloadActivity) Execute(ctx context.Context, pipelineName, watcherName, key string) (string, error)

type HidePackageActivity

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

func NewHidePackageActivity

func NewHidePackageActivity(m *manager.Manager) *HidePackageActivity

func (*HidePackageActivity) Execute

func (a *HidePackageActivity) Execute(ctx context.Context, unitID, unitType, pipelineName string) error

type PollIngestActivity

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

func NewPollIngestActivity

func NewPollIngestActivity(pipelineRegistry *pipeline.Registry) *PollIngestActivity

func (*PollIngestActivity) Execute

func (a *PollIngestActivity) Execute(ctx context.Context, params *PollIngestActivityParams) (time.Time, error)

type PollIngestActivityParams

type PollIngestActivityParams struct {
	PipelineName string
	SIPID        string
}

type PollTransferActivity

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

PollTransferActivity polls the Transfer Status API repeatedly until processing completes or when an error is considered non permanent.

It is expected to deliver at least on heartbeat per minute.

func NewPollTransferActivity

func NewPollTransferActivity(pipelineRegistry *pipeline.Registry) *PollTransferActivity

func (*PollTransferActivity) Execute

func (a *PollTransferActivity) Execute(ctx context.Context, params *PollTransferActivityParams) (string, error)

type PollTransferActivityParams

type PollTransferActivityParams struct {
	PipelineName string
	TransferID   string
}

type PopulateMetadataActivity

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

PopulateMetadataActivity populates the DC identifier in the transfer metadata CSV document.

func NewPopulateMetadataActivity

func NewPopulateMetadataActivity(pipelineRegistry *pipeline.Registry) *PopulateMetadataActivity

func (*PopulateMetadataActivity) Execute

func (a *PopulateMetadataActivity) Execute(ctx context.Context, params *PopulateMetadataActivityParams) error

type PopulateMetadataActivityParams

type PopulateMetadataActivityParams struct {
	Path       string
	Identifier string
}

type TransferActivity

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

TransferActivity submits the transfer to Archivematica and returns its ID.

This is our first interaction with Archivematica. The workflow ends here after authentication errors.

func NewTransferActivity

func NewTransferActivity(m *manager.Manager) *TransferActivity

func (*TransferActivity) Execute

func (a *TransferActivity) Execute(ctx context.Context, params *TransferActivityParams) (*TransferActivityResponse, error)

type TransferActivityParams

type TransferActivityParams struct {
	PipelineName       string
	TransferLocationID string
	RelPath            string
	Name               string
	ProcessingConfig   string
	TransferType       string
	Accession          string
}

type TransferActivityResponse

type TransferActivityResponse struct {
	TransferID      string
	PipelineVersion string
	PipelineID      string
}

type ValidateTransferActivity

type ValidateTransferActivity struct{}

func NewValidateTransferActivity

func NewValidateTransferActivity() *ValidateTransferActivity

func (*ValidateTransferActivity) Execute

func (a *ValidateTransferActivity) Execute(ctx context.Context, params *ValidateTransferActivityParams) error

type ValidateTransferActivityParams

type ValidateTransferActivityParams struct {
	Config validation.Config
	Path   string
}