Support for Archives(zip,tar)

internal/pkg/pipeline/task/archive/archive.go

@@ -0,0 +1,111 @@
+package archive
+
+import (
+	"fmt"
+
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/record"
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task"
+)
+
+type actionType string
+
+const (
+	actionCompress   actionType = `pack`
+	actionDecompress actionType = `unpack`
+)
+
+const (
+	defaultFormat = `zip`
+	defaultAction = `pack`
+)
+
+type archiver interface {
+	Read(b []byte)
+	Write(b []byte)
+}
+
+type core struct {
+	task.Base `yaml:",inline" json:",inline"`
+	Format    string     `yaml:"format,omitempty" json:"format,omitempty"`
+	Action    actionType `yaml:"action,omitempty" json:"action,omitempty"`
+	FileName  string     `yaml:"file_name,omitempty" json:"file_name,omitempty"`
+}
+
+func New() (task.Task, error) {
+	return &core{
+		Format: defaultFormat,
+		Action: defaultAction,
+	}, nil
+}
+
+func (c *core) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	type raw core
+	obj := raw{
+		Format: defaultFormat,
+		Action: defaultAction,
+	}
+	if err := unmarshal(&obj); err != nil {
+		return err
+	}
+
+	if obj.Action != actionCompress && obj.Action != actionDecompress {
+		return fmt.Errorf("invalid action: %s (must be 'compress' or 'decompress')", obj.Action)
+	}
+
+	if obj.Action == actionCompress {
+		if obj.FileName == "" {
+			return fmt.Errorf("file_name must be specified when action is 'pack'")
+		}
+	}
+
+	*c = core(obj)
+
+	return nil
+}
+
+func (c *core) Run(input <-chan *record.Record, output chan<- *record.Record) (err error) {
+
+	if input == nil {
+		return task.ErrNilInput
+	}
+
+	for {
+		r, ok := c.GetRecord(input)
+		if !ok {
+			break
+		}
+
+		if len(r.Data) == 0 {
+			continue
+		}
+
+		var archiv archiver
+
+		switch c.Format {
+		case "tar":
+			archiv = &tarArchive{
+				Base:       &c.Base,
+				FileName:   c.FileName,
+				Record:     r,
+				OutputChan: output,
+			}
+		case "zip":
+			archiv = &zipArchive{
+				Base:       &c.Base,
+				FileName:   c.FileName,
+				Record:     r,
+				OutputChan: output,
+			}
+		default:
+			return fmt.Errorf("unsupported format: %s", c.Format)
+		}
+
+		switch c.Action {
+		case actionCompress:
+			archiv.Write(r.Data)
+		case actionDecompress:
+			archiv.Read(r.Data)
+		}
+	}
+	return nil
+}

internal/pkg/pipeline/task/archive/tar.go

@@ -0,0 +1,65 @@
+package archive
+
+import (
+	"archive/tar"
+	"bytes"
+	"io"
+	"log"
+
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/record"
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task"
+)
+
+type tarArchive struct {
+	*task.Base
+	FileName   string
+	Record     *record.Record
+	OutputChan chan<- *record.Record
+}
+
+func (t *tarArchive) Read(b []byte) {
+	r := tar.NewReader(bytes.NewReader(b))
+
+	for {
+		header, err := r.Next()
+		if err != nil {
+			break
+		}
+
+		// check the file type is regular file
+		if header.Typeflag == tar.TypeReg {
+			buf := make([]byte, header.Size)
+			if _, err := io.ReadFull(r, buf); err != nil && err != io.EOF {
+				log.Fatal(err)
+			}
+			t.SendData(t.Record.Context, buf, t.OutputChan)
+		}
+
+	}
+}
+
+func (t *tarArchive) Write(b []byte) {
+
+	if t.FileName == "" {
+		log.Fatal("file name is required to create tar archive")
+	}
+
+	tw := tar.NewWriter(bytes.NewBuffer(b))
+
+	tw.WriteHeader(
+		&tar.Header{Name: t.FileName,
+			Mode: 0777,
+			Size: int64(len(b))})
+
+	_, err := tw.Write(b)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = tw.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	t.SendData(t.Record.Context, b, t.OutputChan)
+}

internal/pkg/pipeline/task/archive/zip.go

@@ -0,0 +1,66 @@
+package archive
+
+import (
+	"archive/zip"
+	"bytes"
+	"errors"
+	"io"
+	"log"
+
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/record"
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task"
+)
+
+type zipArchive struct {
+	*task.Base
+	FileName   string
+	Record     *record.Record
+	OutputChan chan<- *record.Record
+}
+
+func (z *zipArchive) Read(b []byte) {
+	r, err := zip.NewReader(bytes.NewReader(b), int64(len(b)))
+	if err != nil {
+		log.Fatal(err)
+	}
+	for _, f := range r.File {
+
+		// check the file type is regular file
+		if f.FileInfo().Mode().IsRegular() {
+
+			rc, err := f.Open()
+			if err != nil {
+				log.Fatal(err)
+			}
+
+			buf := make([]byte, f.FileInfo().Size())
+
+			_, err = rc.Read(buf)
+			if err != nil && !errors.Is(err, io.EOF) {
+				log.Fatal(err)
+			}
+
+			rc.Close()
+
+			z.SendData(z.Record.Context, buf, z.OutputChan)
+		}
+	}
+}
+
+func (z *zipArchive) Write(b []byte) {
+
+	zipBuf := new(bytes.Buffer)
+	zipWriter := zip.NewWriter(zipBuf)
+
+	if z.FileName == "" {
+		log.Fatal("file name is required to create zip archive")
+	}
+
+	w, _ := zipWriter.Create(z.FileName)
+	w.Write(b)
+
+	zipWriter.Close()
+
+	z.SendData(z.Record.Context, zipBuf.Bytes(), z.OutputChan)
+
+}

internal/pkg/pipeline/task/compress/formats.go

@@ -26,7 +26,7 @@ var (
 				return io.NopCloser(snappy.NewReader(r)), nil
 			},
 			NewWriter: func(w io.Writer) io.WriteCloser {
-				return snappy.NewWriter(w)
+				return snappy.NewBufferedWriter(w)
 			}},
 	}
 )

internal/pkg/pipeline/tasks.go

@@ -7,6 +7,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task"
+	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task/archive"
 	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task/aws/parameter_store"
 	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task/compress"
 	"github.com/patterninc/caterpillar/internal/pkg/pipeline/task/converter"
@@ -33,6 +34,7 @@ type tasks []task.Task
 var (
 	validate       = validator.New()
 	supportedTasks = map[string]func() (task.Task, error){
+		`archive`:             archive.New,
 		`aws_parameter_store`: parameter_store.New,
 		`compress`:            compress.New,
 		`converter`:           converter.New,

test/pipelines/birds.txt

@@ -0,0 +1,124 @@
+Albatross
+Acorn Woodpecker
+American Kestrel
+Anna's Hummingbird
+Bald Eagle
+Baltimore Oriole
+Barn Swallow
+Belted Kingfisher
+Bicolored Antbird
+Black Capped Chickadee
+Black Skimmer
+Blue Jay
+Bluebird
+Bobolink
+Bohemian Waxwing
+Brown Creeper
+Brown Pelican
+Burrowing Owl
+California Condor
+California Quail
+Canada Goose
+Cardinal
+Caspian Tern
+Cedar Waxwing
+Chestnut Sided Warbler
+Chimney Swift
+Chipping Sparrow
+Clark's Nutcracker
+Clay Colored Sparrow
+Cliff Swallow
+Columbiformes
+Common Eider
+Common Goldeneye
+Common Grackle
+Common Loon
+Common Merganser
+Common Raven
+Common Tern
+Common Yellowthroat
+Coopers Hawk
+Cory's Shearwater
+Crested Flycatcher
+Curve Billed Thrasher
+Dark Eyed Junco
+Dickcissel
+Dovekie
+Downy Woodpecker
+Drab Seedeater
+Dunnock
+Eastern Bluebird
+Eastern Meadowlark
+Eastern Phoebe
+Eastern Screech Owl
+Eastern Towhee
+Eastern Wood Pewee
+Eared Grebe
+Egyptian Plover
+Elanus leucurus
+Evening Grosbeak
+Eared Quetzal
+Eurasian Wigeon
+Eurpean Starling
+Fabulous Flamingo
+Ferruginous Hawk
+Fiscal Flycatcher
+Flammulated Owl
+Flatbill
+Flesh Footed Shearwater
+Florida Jay
+Fringilla coelebs
+Fulmar
+Gadwall
+Gambel's Quail
+Gannet
+Garden Warbler
+Gnatcatcher
+Godwit
+Golden Eagle
+Golden Winged Warbler
+Goldeneye
+Goldfinch
+Goosander
+Goshawk
+Grace's Warbler
+Grasshopper Sparrow
+Gray Catbird
+Great Black Backed Gull
+Great Blue Heron
+Great Crested Flycatcher
+Great Horned Owl
+Great Kiskadee
+Great Spotted Woodpecker
+Great Tit
+Grebe
+Greenbul
+Green Heron
+Green Tailed Towhee
+Green Winged Teal
+Greenlet
+Grey Kingbird
+Grey Owl
+Grosbeaks
+Grouse
+Gull
+Hairy Woodpecker
+Hammond's Flycatcher
+Harris Hawk
+Harris Sparrow
+Hawaiian Creeper
+Hawaiian Goose
+Hawfinch
+Heathland Francolin
+Herring Gull
+Hoary Puffleg
+Hooded Merganser
+Hooded Oriole
+Hooded Warbler
+Hoopoe
+Horned Auk
+Horned Grebe
+Horned Lark
+House Finch
+House Sparrow
+House Wren