108 lines
2.3 KiB
Go
108 lines
2.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
stdHttp "net/http"
|
|
_ "net/http/pprof"
|
|
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
"github.com/ThreeDotsLabs/watermill-http/pkg/http"
|
|
"github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
|
|
"github.com/ThreeDotsLabs/watermill/message"
|
|
"github.com/ThreeDotsLabs/watermill/message/router/middleware"
|
|
"github.com/ThreeDotsLabs/watermill/message/router/plugin"
|
|
)
|
|
|
|
var (
|
|
kafkaAddr = flag.String("kafka", "localhost:9092", "The address of the kafka broker")
|
|
httpAddr = flag.String("http", ":8080", "The address for the http subscriber")
|
|
)
|
|
|
|
type Webhook struct {
|
|
ObjectKind string `json:"object_kind"`
|
|
}
|
|
|
|
func main() {
|
|
flag.Parse()
|
|
logger := watermill.NewStdLogger(true, true)
|
|
|
|
kafkaPublisher, err := kafka.NewPublisher(
|
|
kafka.PublisherConfig{
|
|
Brokers: []string{*kafkaAddr},
|
|
Marshaler: kafka.DefaultMarshaler{},
|
|
},
|
|
logger,
|
|
)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
httpSubscriber, err := http.NewSubscriber(
|
|
*httpAddr,
|
|
http.SubscriberConfig{
|
|
UnmarshalMessageFunc: func(topic string, request *stdHttp.Request) (*message.Message, error) {
|
|
b, err := io.ReadAll(request.Body)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot read body: %w", err)
|
|
}
|
|
|
|
return message.NewMessage(watermill.NewUUID(), b), nil
|
|
},
|
|
},
|
|
logger,
|
|
)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
r, err := message.NewRouter(
|
|
message.RouterConfig{},
|
|
logger,
|
|
)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
r.AddMiddleware(
|
|
middleware.Recoverer,
|
|
middleware.CorrelationID,
|
|
)
|
|
r.AddPlugin(plugin.SignalsHandler)
|
|
|
|
r.AddHandler(
|
|
"http_to_kafka",
|
|
"/webhooks", // this is the URL of our API
|
|
httpSubscriber,
|
|
"webhooks", // this is the topic the message will be published to
|
|
kafkaPublisher,
|
|
func(msg *message.Message) ([]*message.Message, error) {
|
|
webhook := Webhook{}
|
|
|
|
if err := json.Unmarshal(msg.Payload, &webhook); err != nil {
|
|
return nil, fmt.Errorf("cannot unmarshal message: %w", err)
|
|
}
|
|
|
|
// Add simple validation
|
|
if webhook.ObjectKind == "" {
|
|
return nil, errors.New("empty object kind")
|
|
}
|
|
|
|
// Simply forward the message from HTTP Subscriber to Kafka Publisher
|
|
return []*message.Message{msg}, nil
|
|
},
|
|
)
|
|
|
|
go func() {
|
|
// HTTP server needs to be started after the router is ready.
|
|
<-r.Running()
|
|
_ = httpSubscriber.StartHTTPServer()
|
|
}()
|
|
|
|
_ = r.Run(context.Background())
|
|
}
|