vikunja/pkg/events/events.go

// Vikunja is a to-do list application to facilitate your life.
// Copyright 2018-2021 Vikunja and contributors. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public Licensee as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public Licensee for more details.
//
// You should have received a copy of the GNU Affero General Public Licensee
// along with this program.  If not, see <https://www.gnu.org/licenses/>.

package events

import (
	"context"
	"encoding/json"
	"time"

	"code.vikunja.io/api/pkg/log"
	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
)

var pubsub *gochannel.GoChannel

type Event interface {
	TopicName() string
	Message() interface{}
}

func InitEvents() (err error) {
	logger := log.NewWatermillLogger()

	router, err := message.NewRouter(
		message.RouterConfig{},
		logger,
	)
	if err != nil {
		return err
	}

	router.AddMiddleware(
		middleware.Retry{
			MaxRetries:      5,
			InitialInterval: time.Millisecond * 100,
			Logger:          logger,
			Multiplier:      2,
		}.Middleware,
		middleware.Recoverer,
	)

	pubsub = gochannel.NewGoChannel(
		gochannel.Config{
			OutputChannelBuffer: 1024,
		},
		logger,
	)

	for topic, funcs := range listeners {
		for _, handler := range funcs {
			router.AddNoPublisherHandler(topic+watermill.NewShortUUID(), topic, pubsub, func(msg *message.Message) error {
				return handler.Handle(msg.Payload)
			})
		}
	}

	return router.Run(context.Background())
}

func Publish(event Event) error {
	content, err := json.Marshal(event.Message())
	if err != nil {
		return err
	}

	msg := message.NewMessage(watermill.NewUUID(), content)
	return pubsub.Publish(event.TopicName(), msg)
}