Stream create file actions to Kinesis

handlers/deposit_file.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"encoding/json"
 	"log"
 
 	"github.com/go-openapi/runtime"
@@ -10,6 +11,7 @@ import (
 	"github.com/sul-dlss-labs/taco/generated/restapi/operations"
 	"github.com/sul-dlss-labs/taco/identifier"
 	"github.com/sul-dlss-labs/taco/storage"
+	"github.com/sul-dlss-labs/taco/streaming"
 	"github.com/sul-dlss-labs/taco/uploaded"
 	"github.com/sul-dlss-labs/taco/validators"
 )
@@ -18,15 +20,19 @@ const atContext = "http://sdr.sul.stanford.edu/contexts/taco-base.jsonld"
 const fileType = "http://sdr.sul.stanford.edu/contexts/sdr3-file.jsonld"
 
 // NewDepositFile -- Accepts requests to create a file and pushes it to s3.
-func NewDepositFile(database db.Database, uploader storage.Storage, identifierService identifier.Service) operations.DepositFileHandler {
-	return &depositFileEntry{database: database,
+func NewDepositFile(database db.Database, uploader storage.Storage, stream streaming.Stream, identifierService identifier.Service) operations.DepositFileHandler {
+	return &depositFileEntry{
+		database:          database,
 		storage:           uploader,
-		identifierService: identifierService}
+		stream:            stream,
+		identifierService: identifierService,
+	}
 }
 
 type depositFileEntry struct {
 	database          db.Database
 	storage           storage.Storage
+	stream            streaming.Stream
 	identifierService identifier.Service
 }
 
@@ -52,6 +58,11 @@ func (d *depositFileEntry) Handle(params operations.DepositFileParams) middlewar
 	if err := d.createFileResource(id, params.Upload.Header.Filename); err != nil {
 		panic(err)
 	}
+
+	if err := d.addToStream(&id); err != nil {
+		panic(err)
+	}
+
 	// TODO: return file location: https://github.com/sul-dlss-labs/taco/issues/160
 	response := map[string]interface{}{"id": id}
 	return operations.NewDepositResourceCreated().WithPayload(response)
@@ -71,6 +82,15 @@ func (d *depositFileEntry) createFileResource(resourceID string, filename string
 	return d.database.Insert(resource)
 }
 
+func (d *depositFileEntry) addToStream(id *string) error {
+	message, err := json.Marshal(id)
+	if err != nil {
+		return err
+	}
+
+	return d.stream.SendMessage(string(message))
+}
+
 func (d *depositFileEntry) buildPersistableResource(resourceID string, filename string) datautils.Resource {
 	// TODO: Expand here if we need to set any default properties on the file
 	identification := map[string]interface{}{"filename": filename, "identifier": resourceID, "sdrUUID": resourceID}

handlers/deposit_file_test.go

@@ -22,20 +22,23 @@ func TestCreateFileHappyPath(t *testing.T) {
 	r := gofight.New()
 	storage := NewMockStorage()
 	repo := NewMockDatabase(nil)
+	stream := NewMockStream()
 
 	r.POST(filePath).
 		SetHeader(gofight.H{
 			"Content-Type": contentType,
 		}).
 		SetBody(body).
-		Run(handler(repo, nil, storage),
+		Run(handler(repo, stream, storage),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusCreated, r.Code)
 				assert.Equal(t, 1, len(storage.(*MockStorage).CreatedFiles))
 				assert.Equal(t, 1, len(repo.(*MockDatabase).CreatedResources))
 				fileResource := repo.(*MockDatabase).CreatedResources[0]
 				fileName := fileResource["identification"].(map[string]interface{})["filename"]
 				assert.Equal(t, fileName, "foo.txt")
+				assert.Equal(t, 1, len(stream.(*MockStream).Messages))
+
 			})
 }
 

handlers/deposit_resource.go

@@ -2,7 +2,6 @@ package handlers
 
 import (
 	"encoding/json"
-	"log"
 
 	"github.com/go-openapi/runtime/middleware"
 	"github.com/sul-dlss-labs/taco/datautils"
@@ -71,8 +70,6 @@ func (d *depositResource) addToStream(id *string) error {
 	if err != nil {
 		return err
 	}
-	if d.stream == nil {
-		log.Printf("Stream is nil")
-	}
+
 	return d.stream.SendMessage(string(message))
 }

handlers/deposit_resource_test.go

@@ -26,11 +26,10 @@ func postData() map[string]interface{} {
 func TestCreateResourceHappyPath(t *testing.T) {
 	r := gofight.New()
 	repo := NewMockDatabase(nil)
-	stream := NewMockStream("")
 
 	r.POST("/v1/resource").
 		SetJSON(postData()).
-		Run(handler(repo, stream, nil),
+		Run(handler(repo, nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusCreated, r.Code)
 				assert.Equal(t, 1, len(repo.(*MockDatabase).CreatedResources))
@@ -45,7 +44,7 @@ func TestCreateResourceMissingSourceId(t *testing.T) {
 			"id":    "oo000oo0001",
 			"title": "My work",
 		}).
-		Run(handler(NewMockDatabase(nil), NewMockStream(""), nil),
+		Run(handler(nil, nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusUnprocessableEntity, r.Code)
 			})
@@ -63,7 +62,7 @@ func TestCreateResourceSemanticallyValid(t *testing.T) {
 			"preserve": true,
 			"publish":  true,
 			"sourceId": "bib12345678"}).
-		Run(handler(NewMockDatabase(nil), NewMockStream(""), nil),
+		Run(handler(NewMockDatabase(nil), nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusUnprocessableEntity, r.Code)
 			})
@@ -75,7 +74,7 @@ func TestCreateResourceFailure(t *testing.T) {
 		func() {
 			r.POST("/v1/resource").
 				SetJSON(postData()).
-				Run(handler(NewMockErrorDatabase(), NewMockStream(""), nil),
+				Run(handler(NewMockErrorDatabase(), nil, nil),
 					func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {})
 		})
 }

handlers/health_check_test.go

@@ -12,7 +12,7 @@ import (
 func TestHealthCheck(t *testing.T) {
 	r := gofight.New()
 	r.GET("/v1/healthcheck").
-		Run(handler(NewMockDatabase(nil), NewMockStream(""), nil),
+		Run(handler(nil, nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusOK, r.Code)
 				stat, _ := jsonparser.GetString(r.Body.Bytes(), "status")

handlers/server.go

@@ -21,7 +21,7 @@ func BuildAPI(database db.Database, stream streaming.Stream, storage storage.Sto
 	api.DepositResourceHandler = NewDepositResource(database, stream, depositValidator, identifierService)
 	updateValidator := validators.NewUpdateResourceValidator(database)
 	api.UpdateResourceHandler = NewUpdateResource(database, stream, updateValidator)
-	api.DepositFileHandler = NewDepositFile(database, storage, identifierService)
+	api.DepositFileHandler = NewDepositFile(database, storage, stream, identifierService)
 	api.HealthCheckHandler = NewHealthCheck()
 	return api
 }

handlers/stub_database_test.go

@@ -16,7 +16,7 @@ func handler(database db.Database, stream streaming.Stream, storage storage.Stor
 		database = NewMockDatabase(nil)
 	}
 	if stream == nil {
-		stream = NewMockStream("")
+		stream = NewMockStream()
 	}
 	if storage == nil {
 		storage = NewMockStorage()

handlers/stub_stream_test.go

@@ -3,13 +3,14 @@ package handlers
 import "github.com/sul-dlss-labs/taco/streaming"
 
 type MockStream struct {
-	message string
+	Messages []string
 }
 
-func NewMockStream(message string) streaming.Stream {
-	return &MockStream{message: message}
+func NewMockStream() streaming.Stream {
+	return &MockStream{Messages: []string{}}
 }
 
-func (d *MockStream) SendMessage(message string) error {
+func (s *MockStream) SendMessage(message string) error {
+	s.Messages = append(s.Messages, message)
 	return nil
 }

handlers/update_resource_test.go

@@ -44,7 +44,7 @@ func TestUpdateResourceHappyPath(t *testing.T) {
 	r.PATCH("/v1/resource/99").
 		SetHeader(gofight.H{"Content-Type": "application/json"}).
 		SetJSON(updateMessage).
-		Run(handler(repo, NewMockStream(""), nil),
+		Run(handler(repo, nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusOK, r.Code)
 			})
@@ -55,7 +55,7 @@ func TestUpdateResourceNotFound(t *testing.T) {
 	r.PATCH("/v1/resource/99").
 		SetHeader(gofight.H{"Content-Type": "application/json"}).
 		SetJSON(updateMessage).
-		Run(handler(NewMockDatabase(nil), NewMockStream(""), nil),
+		Run(handler(nil, nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusNotFound, r.Code)
 			})
@@ -64,7 +64,7 @@ func TestUpdateResourceNotFound(t *testing.T) {
 func TestUpdateResourceEmptyRequest(t *testing.T) {
 	r := gofight.New()
 	r.PATCH("/v1/resource/100").
-		Run(handler(NewMockDatabase(nil), NewMockStream(""), nil),
+		Run(handler(nil, nil, nil),
 			func(r gofight.HTTPResponse, rq gofight.HTTPRequest) {
 				assert.Equal(t, http.StatusUnprocessableEntity, r.Code)
 			})