Merge 95dc207 into 577de21

Dockerfile

@@ -0,0 +1,21 @@
+FROM golang as builder
+
+RUN apt-get update && apt-get install -y bash curl git go-dep libmagickwand-6.q16-dev
+WORKDIR /go/src/github.com/analogj/lodestone-processor/
+COPY . /go/src/github.com/analogj/lodestone-processor/
+COPY 152-2009-R0911_A1.pdf /tmp/152-2009-R0911_A1.pdf
+RUN dep ensure && \
+    go build -o processor ./cmd/main.go && \
+    ./processor
+
+#GOOS=linux GARCH=amd64 CGO_ENABLED=0 go build -a -installsuffix cgo -ldflags "-X main.goos=linux -X main.goarch=amd64 -extldflags \"-static\"" -o lodestone-document-processor-linux-amd64 ./cmd/document-processor/document-processor.go && \
+#GOOS=linux GARCH=amd64 CGO_ENABLED=0 go build -a -installsuffix cgo -ldflags "-X main.goos=linux -X main.goarch=amd64 -extldflags \"-static\"" -o lodestone-thumbnail-processor-linux-amd64 ./cmd/thumbnail-processor/thumbnail-processor.go
+#
+#
+#FROM debian:buster-slim as runtime
+#
+#RUN apt-get update && apt-get install -y bash curl git go-dep libmagickwand-6.q16-dev libreoffice-common
+#COPY --from=builder /go/src/github.com/analogj/lodestone-processor/lodestone-document-processor-linux-amd64 /usr/bin/lodestone-document-processor
+#COPY --from=builder /go/src/github.com/analogj/lodestone-processor/lodestone-thumbnail-processor-linux-amd64 /usr/bin/lodestone-thumbnail-processor
+#
+#RUN chmod +x /usr/bin/lodestone-document-processor /usr/bin/lodestone-thumbnail-processor

Gopkg.toml

@@ -38,5 +38,5 @@
   name = "github.com/analogj/go-util"
 
 [[constraint]]
-  version = "=v6.15.5"
-  name = "github.com/go-redis/redis"
+  name = "gopkg.in/gographics/imagick.v2"
+  version = "2.5.0"

capsule.yml

@@ -2,8 +2,9 @@
 engine_enable_code_mutation: true
 
 engine_cmd_compile:
-  - 'GOOS=linux GARCH=amd64 CGO_ENABLED=0 go build -a -installsuffix cgo -ldflags "-X main.goos=linux -X main.goarch=amd64 -extldflags \"-static\"" -o lodestone-document-processor-linux-amd64 ./cmd/document-processor/document-processor.go'
-  - 'GOOS=linux GARCH=amd64 CGO_ENABLED=0 go build -a -installsuffix cgo -ldflags "-X main.goos=linux -X main.goarch=amd64 -extldflags \"-static\"" -o lodestone-thumbnail-processor-linux-amd64 ./cmd/thumbnail-processor/thumbnail-processor.go'
+  - 'apt-get update && apt-get install -y bash curl git go-dep libmagickwand-6.q16-dev libreoffice-common'
+  - 'go build -o lodestone-document-processor-linux-amd64 ./cmd/document-processor/document-processor.go'
+  - 'go build -o lodestone-thumbnail-processor-linux-amd64 ./cmd/thumbnail-processor/thumbnail-processor.go'
 engine_cmd_test: 'go test -v ./...'
 engine_disable_lint: true
 

cmd/document-processor/document-processor.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/analogj/go-util/utils"
 	"github.com/analogj/lodestone-processor/pkg/listen"
+	"github.com/analogj/lodestone-processor/pkg/processor"
 	"github.com/analogj/lodestone-processor/pkg/version"
 	"github.com/fatih/color"
 	"github.com/urfave/cli"
@@ -56,19 +57,25 @@ func main() {
 
 					listenClient = new(listen.AmqpListen)
 					err := listenClient.Init(map[string]string{
-						"amqp-url": c.String("amqp-url"),
-						"exchange": c.String("amqp-exchange"),
-						"queue":    c.String("amqp-queue"),
+						"amqp-url":    c.String("amqp-url"),
+						"exchange":    c.String("amqp-exchange"),
+						"queue":       c.String("amqp-queue"),
+						"storage-url": c.String("storage-url"),
 					})
 					if err != nil {
 						return err
 					}
 					defer listenClient.Close()
 
-					return listenClient.Subscribe()
+					return listenClient.Subscribe(processor.DocumentProcessor)
 				},
 
 				Flags: []cli.Flag{
+					&cli.StringFlag{
+						Name:  "storage-url",
+						Usage: "The storage server url",
+						Value: "http://storage:9000",
+					},
 
 					&cli.StringFlag{
 						Name:  "amqp-url",

cmd/thumbnail-processor/thumbnail-processor.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/analogj/go-util/utils"
 	"github.com/analogj/lodestone-processor/pkg/listen"
+	"github.com/analogj/lodestone-processor/pkg/processor"
 	"github.com/analogj/lodestone-processor/pkg/version"
 	"github.com/fatih/color"
 	"github.com/urfave/cli"
@@ -56,20 +57,27 @@ func main() {
 
 					listenClient = new(listen.AmqpListen)
 					err := listenClient.Init(map[string]string{
-						"amqp-url": c.String("amqp-url"),
-						"exchange": c.String("amqp-exchange"),
-						"queue":    c.String("amqp-queue"),
+						"amqp-url":    c.String("amqp-url"),
+						"exchange":    c.String("amqp-exchange"),
+						"queue":       c.String("amqp-queue"),
+						"storage-url": c.String("storage-url"),
 					})
 					if err != nil {
 						return err
 					}
 					defer listenClient.Close()
 
-					return listenClient.Subscribe()
+					return listenClient.Subscribe(processor.ThumbnailProcessor)
 				},
 
 				Flags: []cli.Flag{
 
+					&cli.StringFlag{
+						Name:  "storage-url",
+						Usage: "The storage server url",
+						Value: "http://storage:9000",
+					},
+
 					&cli.StringFlag{
 						Name:  "amqp-url",
 						Usage: "The amqp connection string",

pkg/listen/amqp.go

@@ -7,15 +7,17 @@ import (
 )
 
 type AmqpListen struct {
-	client   *amqp.Connection
-	channel  *amqp.Channel
-	exchange string
-	queue    string
+	client     *amqp.Connection
+	channel    *amqp.Channel
+	exchange   string
+	queue      string
+	storageUrl string
 }
 
 func (n *AmqpListen) Init(config map[string]string) error {
 	n.exchange = config["exchange"]
 	n.queue = config["queue"]
+	n.storageUrl = config["storage-url"]
 
 	client, err := amqp.Dial(config["amqp-url"])
 	if err != nil {
@@ -58,7 +60,7 @@ func (n *AmqpListen) Init(config map[string]string) error {
 	return err
 }
 
-func (n *AmqpListen) Subscribe() error {
+func (n *AmqpListen) Subscribe(processor func(body []byte, storageUrl string) error) error {
 	fmt.Println("Subscribe to events..")
 
 	err := n.channel.QueueBind(
@@ -89,6 +91,10 @@ func (n *AmqpListen) Subscribe() error {
 	go func() {
 		for d := range msgs {
 			log.Printf(" [x] %s", d.Body)
+			if err := processor(d.Body, n.storageUrl); err != nil {
+				log.Printf("Error when processing document %s", err)
+				//TODO: add to the dead letter queue (for further processing later)
+			}
 		}
 	}()
 

pkg/listen/interface.go

@@ -2,6 +2,6 @@ package listen
 
 type Interface interface {
 	Init(config map[string]string) error
-	Subscribe() error
+	Subscribe(processor func(body []byte, storageUrl string) error) error
 	Close() error
 }

pkg/processor/document.go

@@ -0,0 +1,8 @@
+package processor
+
+func DocumentProcessor(body []byte, storageUrl string) error {
+
+	//TODO pass document to TIKA
+
+	return nil
+}