use a prefix for files generated on s3

s3_emitter.go

@@ -18,13 +18,18 @@ import (
 // dash. This struct requires the configuration of an S3 bucket and endpoint.
 type S3Emitter struct {
 	S3Bucket string
+	S3Prefix string
 }
 
 // S3FileName generates a file name based on the First and Last sequence numbers from the buffer. The current
 // UTC date (YYYY-MM-DD) is base of the path to logically group days of batches.
 func (e S3Emitter) S3FileName(firstSeq string, lastSeq string) string {
 	date := time.Now().UTC().Format("2006/01/02")
-	return fmt.Sprintf("%v/%v-%v", date, firstSeq, lastSeq)
+	if e.S3Prefix == "" {
+		return fmt.Sprintf("%v/%v-%v", date, firstSeq, lastSeq)
+	} else {
+		return fmt.Sprintf("%v/%v/%v-%v", e.S3Prefix, date, firstSeq, lastSeq)
+	}
 }
 
 // Emit is invoked when the buffer is full. This method emits the set of filtered records.

s3_emitter_test.go

@@ -8,12 +8,22 @@ import (
 
 func TestS3FileName(t *testing.T) {
 	d := time.Now().UTC().Format("2006/01/02")
-	e := S3Emitter{}
+	e := S3Emitter{S3Bucket: "bucket", S3Prefix: "prefix"}
 
-	expected := fmt.Sprintf("%v/a-b", d)
+	expected := fmt.Sprintf("prefix/%v/a-b", d)
 	result := e.S3FileName("a", "b")
 
 	if result != expected {
 		t.Errorf("S3FileName() = %v want %v", result, expected)
 	}
+
+	e.S3Prefix = ""
+
+	expected = fmt.Sprintf("%v/a-b", d)
+	result = e.S3FileName("a", "b")
+
+	if result != expected {
+		t.Errorf("S3FileName() = %v want %v", result, expected)
+	}
+
 }