created builder trigger hook, added monitoring, added delete method to the proto

Author: vlad

Dockerfile

@@ -0,0 +1,9 @@
+FROM golang:1.15
+
+WORKDIR /app
+
+COPY ./ /app
+
+RUN go mod download
+
+ENTRYPOINT go run main.go

README.md

@@ -1,3 +1,3 @@
 `cd proto`
 
-`protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative task.proto`
+`protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative ./proto/task.proto`

go.mod

@@ -4,11 +4,16 @@ go 1.15
 
 require (
 	github.com/golang/protobuf v1.4.3
-	github.com/pvelx/triggerHook v0.0.0-20201201162513-7a80e07e5224
+	github.com/influxdata/influxdb v1.8.3
+	github.com/pkg/errors v0.9.1
+	github.com/pvelx/triggerHook v0.0.0-00010101000000-000000000000
+	github.com/streadway/amqp v1.0.0
 	golang.org/x/net v0.0.0-20201201195509-5d6afe98e0b7 // indirect
 	golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3 // indirect
 	golang.org/x/text v0.3.4 // indirect
 	google.golang.org/genproto v0.0.0-20201201144952-b05cb90ed32e // indirect
 	google.golang.org/grpc v1.33.2
 	google.golang.org/protobuf v1.25.0
 )
+
+replace github.com/pvelx/triggerHook => ../trigger-hook

go.sum

@@ -1,34 +1,70 @@
 package main
 
 import (
-	"github.com/pvelx/triggerHook"
-	"github.com/pvelx/triggerHookExample/proto"
-	"github.com/pvelx/triggerHookExample/sendingTransport"
-	"github.com/pvelx/triggerHookExample/task_server"
-	"google.golang.org/grpc"
+	"encoding/json"
+	"github.com/pvelx/triggerHook/domain"
+	"github.com/streadway/amqp"
 	"log"
-	"net"
 )
 
-var tasksDeferredService = triggerHook.Default()
-
 const (
 	port = ":50051"
 )
 
 func main() {
-	tasksDeferredService.SetTransport(sendingTransport.NewAmqpTransport())
-	go tasksDeferredService.Run()
 
-	taskServer := task_server.New(tasksDeferredService)
+	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
+	if err != nil {
+		log.Fatalf("Dial: %s", err)
+	}
+	defer conn.Close()
 
-	lis, err := net.Listen("tcp", port)
+	channel, err := conn.Channel()
 	if err != nil {
-		log.Fatalf("failed to listen: %v", err)
+		log.Fatalf("Error open channel:%v", err)
+	}
+
+	transport := func(task domain.Task) {
+		taskJson, err := json.Marshal(task)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		if err = channel.Publish(
+			"test-exchange", // publish to an exchange
+			"",              // routing to 0 or more queues
+			true,            // mandatory
+			false,           // immediate
+			amqp.Publishing{
+				Headers:         amqp.Table{},
+				ContentType:     "application/json",
+				ContentEncoding: "",
+				Body:            taskJson,
+				DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
+				Priority:        0,              // 0-9
+			},
+		); err != nil {
+			log.Fatalf("Exchange Publish: %s", err)
+		}
 	}
-	s := grpc.NewServer()
-	proto.RegisterTaskServer(s, taskServer)
-	if err := s.Serve(lis); err != nil {
+
+	monitoring := NewMonitoring()
+	tasksDeferredService := BuildTriggerHook(monitoring, transport)
+	taskServer := NewTaskServer(tasksDeferredService)
+
+	go func() {
+		if err := tasksDeferredService.Run(); err != nil {
+			log.Fatalf("failed run trigger hook: %v", err)
+		}
+	}()
+
+	go func() {
+		if err := monitoring.Run(); err != nil {
+			log.Fatalf("failed run monitoring: %v", err)
+		}
+	}()
+
+	if err := RunGrpcServer(taskServer); err != nil {
 		log.Fatalf("failed to serve: %v", err)
 	}
 }

main.go

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"github.com/influxdata/influxdb/client/v2"
+	"github.com/pvelx/triggerHook/contracts"
+	"log"
+	"time"
+)
+
+var sampleSize = 1000
+var chPointCap = 10000
+
+type Monitoring struct {
+	connection client.Client
+	chPoint    chan *client.Point
+}
+
+func NewMonitoring() *Monitoring {
+	c, err := client.NewHTTPClient(client.HTTPConfig{
+		Addr:     "http://localhost:8086",
+		Username: "monitor",
+		Password: "secret",
+	})
+	if err != nil {
+		log.Fatalln("Error: ", err)
+	}
+
+	return &Monitoring{
+		connection: c,
+		chPoint:    make(chan *client.Point, chPointCap),
+	}
+}
+
+func (m *Monitoring) Run() error {
+	for {
+		bp, err := client.NewBatchPoints(client.BatchPointsConfig{
+			Database:  "trigger_hook",
+			Precision: "s",
+		})
+		if err != nil {
+			return err
+		}
+		expire := time.After(5 * time.Second)
+		for {
+			select {
+			case point := <-m.chPoint:
+				bp.AddPoint(point)
+
+				if len(bp.Points()) == sampleSize {
+					goto done
+				}
+
+			case <-expire:
+				goto done
+			}
+		}
+
+	done:
+		if len(bp.Points()) > 0 {
+			err = m.connection.Write(bp)
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func (m *Monitoring) AddMeasurement(name string, event contracts.MeasurementEvent) {
+	point, err := client.NewPoint(
+		name,
+		nil,
+		map[string]interface{}{
+			"value": event.Measurement,
+		},
+		event.Time,
+	)
+	if err != nil {
+		log.Fatalln("Error: ", err)
+	}
+
+	m.chPoint <- point
+}