working minio plugin

examples/plugins/minio/go.mod

@@ -5,5 +5,5 @@ go 1.15
 require (
 	github.com/minio/minio-go/v7 v7.0.7
 	github.com/tinyzimmer/go-glib v0.0.18
-	github.com/tinyzimmer/go-gst v0.2.8
+	github.com/tinyzimmer/go-gst v0.2.11
 )

examples/plugins/minio/go.sum

@@ -70,6 +70,8 @@ github.com/tinyzimmer/go-gst v0.2.4 h1:uDGTzObBmIhyukqjCE9Jw0/EmmNU47Ztd5lBrtXTm
 github.com/tinyzimmer/go-gst v0.2.4/go.mod h1:aPV2CtdfNrtASAzj+DzrAISJr1Czfy25ihLJIh7f/tk=
 github.com/tinyzimmer/go-gst v0.2.8 h1:l0O9IjxncP7TMeeDFfYeQjrmsDv4STE0j8gVU1N8J74=
 github.com/tinyzimmer/go-gst v0.2.8/go.mod h1:C1yElEfXm8k0ddR4NdT1cJS4vFHv2wyVrIBSJCB6Nto=
+github.com/tinyzimmer/go-gst v0.2.11 h1:Nfaz7k0L2stRrSGhdGyZbfbbCUMU6/zC0UBi8Ftt8S0=
+github.com/tinyzimmer/go-gst v0.2.11/go.mod h1:C1yElEfXm8k0ddR4NdT1cJS4vFHv2wyVrIBSJCB6Nto=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=

examples/plugins/minio/miniosink.go

@@ -1,14 +1,12 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"os"
 	"strings"
 	"sync"
 
-	minio "github.com/minio/minio-go/v7"
 	"github.com/tinyzimmer/go-glib/glib"
 	"github.com/tinyzimmer/go-gst/gst"
 	"github.com/tinyzimmer/go-gst/gst/base"
@@ -24,12 +22,8 @@ type minioSink struct {
 	settings *settings
 	state    *sinkstate
 
-	rPipe    *io.PipeReader
-	wPipe    *io.PipeWriter
-	doneChan chan struct{}
-	cancel   func()
-
-	mux sync.Mutex
+	writer *seekWriter
+	mux    sync.Mutex
 }
 
 type sinkstate struct {
@@ -119,6 +113,7 @@ func (m *minioSink) Start(self *base.GstBaseSink) bool {
 		m.settings.secretAccessKey = os.Getenv(spl[len(spl)-1])
 	}
 
+	self.Log(sinkCAT, gst.LevelInfo, fmt.Sprintf("Creating new MinIO client for %s", m.settings.endpoint))
 	client, err := getMinIOClient(m.settings)
 	if err != nil {
 		self.Log(sinkCAT, gst.LevelError, err.Error())
@@ -127,31 +122,8 @@ func (m *minioSink) Start(self *base.GstBaseSink) bool {
 		return false
 	}
 
-	m.doneChan = make(chan struct{})
-	m.rPipe, m.wPipe = io.Pipe()
-	var ctx context.Context
-	ctx, m.cancel = context.WithCancel(context.Background())
-
-	go func() {
-		self.Ref()
-		defer self.Unref()
-		defer m.cancel()
-		defer func() { m.doneChan <- struct{}{} }()
-
-		self.Log(sinkCAT, gst.LevelInfo, fmt.Sprintf("Starting PutObject operation to %s/%s/%s", m.settings.endpoint, m.settings.bucket, m.settings.key))
-		info, err := client.PutObject(ctx, m.settings.bucket, m.settings.key, m.rPipe, -1, minio.PutObjectOptions{
-			ContentType: "application/octet-stream",
-			PartSize:    m.settings.partSize,
-		})
-
-		if err != nil {
-			self.ErrorMessage(gst.DomainResource, gst.ResourceErrorFailed,
-				fmt.Sprintf("Failed to do put object request to %s/%s/%s : %s", m.settings.endpoint, m.settings.bucket, m.settings.key, err.Error()), "")
-			return
-		}
-
-		self.Log(sinkCAT, gst.LevelInfo, fmt.Sprintf("PutObject operation has returned: %+v", info))
-	}()
+	self.Log(sinkCAT, gst.LevelInfo, "Initializing new MinIO writer")
+	m.writer = newSeekWriter(client, int64(m.settings.partSize), m.settings.bucket, m.settings.key)
 
 	m.state.started = true
 	self.Log(sinkCAT, gst.LevelInfo, "MinIOSink has started")
@@ -168,18 +140,7 @@ func (m *minioSink) Stop(self *base.GstBaseSink) bool {
 		return false
 	}
 
-	self.Log(sinkCAT, gst.LevelInfo, "Closing write pipe to PutObject operation")
-	if err := m.wPipe.Close(); err != nil {
-		self.Log(sinkCAT, gst.LevelError, err.Error())
-		self.ErrorMessage(gst.DomainResource, gst.ResourceErrorClose, fmt.Sprintf("Failed to finalize MinIO object: %s", err.Error()), "")
-		return false
-	}
-
-	self.Log(sinkCAT, gst.LevelInfo, "Blocking until PutObject operation has completed")
-	<-m.doneChan
-	self.Log(sinkCAT, gst.LevelInfo, "PutObject operation has completed")
-
-	m.rPipe, m.wPipe, m.doneChan, m.cancel = nil, nil, nil, nil
+	m.writer = nil
 	m.state.started = false
 
 	self.Log(sinkCAT, gst.LevelInfo, "MinIOSink has stopped")
@@ -197,7 +158,7 @@ func (m *minioSink) Render(self *base.GstBaseSink, buffer *gst.Buffer) gst.FlowR
 
 	self.Log(sinkCAT, gst.LevelTrace, fmt.Sprintf("Rendering buffer %v", buffer))
 
-	if _, err := m.wPipe.Write(buffer.Bytes()); err != nil {
+	if _, err := m.writer.Write(buffer.Bytes()); err != nil {
 		self.Log(sinkCAT, gst.LevelError, err.Error())
 		self.ErrorMessage(gst.DomainResource, gst.ResourceErrorWrite, fmt.Sprintf("Failed to write data to minio buffer: %s", err.Error()), "")
 		return gst.FlowError
@@ -205,3 +166,49 @@ func (m *minioSink) Render(self *base.GstBaseSink, buffer *gst.Buffer) gst.FlowR
 
 	return gst.FlowOK
 }
+
+func (m *minioSink) Event(self *base.GstBaseSink, event *gst.Event) bool {
+
+	switch event.Type() {
+
+	case gst.EventTypeSegment:
+		segment := event.ParseSegment()
+
+		if segment.GetFormat() == gst.FormatBytes {
+			if uint64(m.writer.currentPosition) != segment.GetStart() {
+				self.Log(sinkCAT, gst.LevelInfo, fmt.Sprintf("Seeking to %d", segment.GetStart()))
+				if _, err := m.writer.Seek(int64(segment.GetStart()), io.SeekStart); err != nil {
+					self.ErrorMessage(gst.DomainResource, gst.ResourceErrorFailed, err.Error(), "")
+				}
+			} else {
+				self.Log(sinkCAT, gst.LevelDebug, "Ignored SEGMENT, no seek needed")
+			}
+		} else {
+			self.Log(sinkCAT, gst.LevelDebug, fmt.Sprintf("Ignored SEGMENT event of format %s", segment.GetFormat().String()))
+		}
+
+	case gst.EventTypeFlushStop:
+		self.Log(sinkCAT, gst.LevelInfo, "Flushing contents of writer and seeking back to start")
+		if m.writer.currentPosition != 0 {
+			if err := m.writer.flush(true); err != nil {
+				self.ErrorMessage(gst.DomainResource, gst.ResourceErrorWrite, err.Error(), "")
+				return false
+			}
+			if _, err := m.writer.Seek(0, io.SeekStart); err != nil {
+				self.ErrorMessage(gst.DomainResource, gst.ResourceErrorFailed, err.Error(), "")
+			}
+		}
+
+	case gst.EventTypeEOS:
+		self.Log(sinkCAT, gst.LevelInfo, "Received EOS, closing MinIO writer")
+		if err := m.writer.Close(); err != nil {
+			self.Log(sinkCAT, gst.LevelError, err.Error())
+			self.ErrorMessage(gst.DomainResource, gst.ResourceErrorClose, fmt.Sprintf("Failed to close MinIO writer: %s", err.Error()), "")
+			return false
+		}
+
+	}
+
+	return self.ParentEvent(event)
+
+}

examples/plugins/minio/seek_writer.go

@@ -0,0 +1,190 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"io/ioutil"
+	"path"
+
+	minio "github.com/minio/minio-go/v7"
+)
+
+type seekWriter struct {
+	// The current position in the buffer
+	currentPosition int64
+	// The size of each part to upload
+	partSize int64
+	// A map of in memory parts to their content
+	parts map[int64][]byte
+	// A map of uploaded parts to the checksum at time of upload
+	uploadedParts map[int64]string
+	// A local reference to the minio client
+	client      *minio.Client
+	bucket, key string
+}
+
+func newSeekWriter(client *minio.Client, partsize int64, bucket, key string) *seekWriter {
+	return &seekWriter{
+		currentPosition: 0,
+		partSize:        partsize,
+		parts:           make(map[int64][]byte),
+		uploadedParts:   make(map[int64]string),
+		client:          client,
+		bucket:          bucket, key: key,
+	}
+}
+
+func (s *seekWriter) Write(p []byte) (int, error) {
+	wrote, err := s.buffer(0, p)
+	if err != nil {
+		return wrote, err
+	}
+	return wrote, s.flush(false)
+}
+
+func (s *seekWriter) Seek(offset int64, whence int) (int64, error) {
+	// Only needs to support SeekStart
+	s.currentPosition = offset
+	return s.currentPosition, nil
+}
+
+func (s *seekWriter) Close() error {
+	if err := s.flush(true); err != nil {
+		return err
+	}
+	if len(s.uploadedParts) == 0 {
+		return nil
+	}
+	opts := make([]minio.CopySrcOptions, len(s.uploadedParts))
+	for i := 0; i < len(opts); i++ {
+		opts[i] = minio.CopySrcOptions{
+			Bucket: s.bucket,
+			Object: s.keyForPart(int64(i)),
+		}
+	}
+	_, err := s.client.ComposeObject(context.Background(), minio.CopyDestOptions{
+		Bucket: s.bucket,
+		Object: s.key,
+	}, opts...)
+
+	if err != nil {
+		return err
+	}
+	for _, opt := range opts {
+		if err := s.client.RemoveObject(context.Background(), opt.Bucket, opt.Object, minio.RemoveObjectOptions{}); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *seekWriter) buffer(from int, p []byte) (int, error) {
+	currentPart := s.currentPosition / s.partSize
+	writeat := s.currentPosition % s.partSize
+	lenToWrite := int64(len(p))
+
+	var buf []byte
+	var ok bool
+	if buf, ok = s.parts[currentPart]; !ok {
+		if _, ok := s.uploadedParts[currentPart]; !ok {
+			s.parts[currentPart] = make([]byte, writeat+lenToWrite)
+			buf = s.parts[currentPart]
+		} else {
+			var err error
+			buf, err = s.fetchRemotePart(currentPart)
+			if err != nil {
+				return from, err
+			}
+		}
+	}
+
+	if lenToWrite+writeat > s.partSize {
+		newbuf := make([]byte, s.partSize)
+		copy(newbuf, buf)
+		s.parts[currentPart] = newbuf
+		buf = newbuf
+	} else if lenToWrite+writeat > int64(len(buf)) {
+		newbuf := make([]byte, lenToWrite+writeat)
+		copy(newbuf, buf)
+		s.parts[currentPart] = newbuf
+		buf = newbuf
+	}
+
+	wrote := copy(buf[writeat:], p)
+
+	s.currentPosition += int64(wrote)
+
+	if int64(wrote) != lenToWrite {
+		return s.buffer(from+wrote, p[wrote:])
+	}
+
+	return from + wrote, nil
+}
+
+func (s *seekWriter) flush(all bool) error {
+	for part, buf := range s.parts {
+		if all || int64(len(buf)) == s.partSize {
+			if err := s.uploadPart(part, buf); err != nil {
+				return err
+			}
+			continue
+		}
+		if !all {
+			continue
+		}
+		if err := s.uploadPart(part, buf); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *seekWriter) uploadPart(part int64, data []byte) error {
+	h := sha256.New()
+	if _, err := h.Write(data); err != nil {
+		return err
+	}
+	datasum := fmt.Sprintf("%x", h.Sum(nil))
+	if sum, ok := s.uploadedParts[part]; ok && sum == datasum {
+		return nil
+	}
+	_, err := s.client.PutObject(context.Background(),
+		s.bucket, s.keyForPart(part),
+		bytes.NewReader(data), int64(len(data)),
+		minio.PutObjectOptions{
+			ContentType: "application/octet-stream",
+		},
+	)
+	if err != nil {
+		return err
+	}
+	delete(s.parts, part)
+	s.uploadedParts[part] = datasum
+	return nil
+}
+
+func (s *seekWriter) fetchRemotePart(part int64) ([]byte, error) {
+	object, err := s.client.GetObject(context.Background(), s.bucket, s.keyForPart(part), minio.GetObjectOptions{})
+	if err != nil {
+		return nil, err
+	}
+	body, err := ioutil.ReadAll(object)
+	if err != nil {
+		return nil, err
+	}
+	s.parts[part] = body
+	return body, nil
+}
+
+func (s *seekWriter) keyForPart(part int64) string {
+	if path.Dir(s.key) == "" {
+		return fmt.Sprintf("%s_tmp/%d", s.key, part)
+	}
+	return path.Join(
+		path.Dir(s.key),
+		fmt.Sprintf("%s_tmp/%d", path.Base(s.key), part),
+	)
+}