Merge pull request #45 from tidepool-org/mailer

Add cloud events mailer

Author: toddkazakov

clients/mailer.go

@@ -0,0 +1,29 @@
+package clients
+
+import (
+	"context"
+	"github.com/tidepool-org/go-common/events"
+)
+
+const MailerTopic = "emails"
+
+type MailerClient interface {
+	SendEmailTemplate(context.Context, events.SendEmailTemplateEvent) error
+}
+
+type mailerClient struct {
+	producer *events.KafkaCloudEventsProducer
+}
+
+func NewMailerClient(config *events.CloudEventsConfig) (MailerClient, error) {
+	producer, err := events.NewKafkaCloudEventsProducer(config)
+	if err != nil {
+		return nil, err
+	}
+
+	return &mailerClient{producer}, nil
+}
+
+func (m *mailerClient) SendEmailTemplate(ctx context.Context, event events.SendEmailTemplateEvent) error {
+	return m.producer.Send(ctx, event)
+}

events/mailer.go

@@ -0,0 +1,54 @@
+package events
+
+import cloudevents "github.com/cloudevents/sdk-go/v2"
+
+const (
+	SendEmailTemplateEventType = "email_template:send"
+)
+
+type SendEmailTemplateEvent struct {
+	Recipient string            `json:"recipient"`
+	Template  string            `json:"template"`
+	Variables map[string]string `json:"variables"`
+}
+
+func (s SendEmailTemplateEvent) GetEventType() string {
+	return SendEmailTemplateEventType
+}
+
+func (s SendEmailTemplateEvent) GetEventKey() string {
+	return s.Recipient
+}
+
+var _ Event = &SendEmailTemplateEvent{}
+
+type EmailEventHandler interface {
+	HandleSendEmailTemplate(payload SendEmailTemplateEvent) error
+}
+
+type DelegatingEmailEventHandler struct {
+	delegate EmailEventHandler
+}
+
+func NewDelegatingEmailEventHandler(delegate EmailEventHandler) *DelegatingEmailEventHandler {
+	return &DelegatingEmailEventHandler{delegate: delegate}
+}
+
+func (d DelegatingEmailEventHandler) CanHandle(ce cloudevents.Event) bool {
+	return ce.Type() == SendEmailTemplateEventType
+}
+
+func (d DelegatingEmailEventHandler) Handle(ce cloudevents.Event) error {
+	if ce.Type() != SendEmailTemplateEventType {
+		// ignore invalid events
+		return nil
+	}
+
+	payload := SendEmailTemplateEvent{}
+	if err := ce.DataAs(&payload); err != nil {
+		return err
+	}
+	return d.delegate.HandleSendEmailTemplate(payload)
+}
+
+var _ EventHandler = &DelegatingEmailEventHandler{}

events/sender.go

@@ -37,6 +37,10 @@ func NewKafkaCloudEventsProducerForDeadLetters(config *CloudEventsConfig) (*Kafk
 }
 
 func newKafkaCloudEventsProducerWithTopic(config *CloudEventsConfig, topic string) (*KafkaCloudEventsProducer, error) {
+	// We are using a sync producer which requires setting the variables below
+	config.SaramaConfig.Producer.Return.Errors = true
+	config.SaramaConfig.Producer.Return.Successes = true
+	
 	sender, err := kafka_sarama.NewSender(config.KafkaBrokers, config.SaramaConfig, topic)
 	if err != nil {
 		return nil, err