discord, and slack platforms added

go.mod

@@ -52,6 +52,7 @@ require (
 	github.com/sha1sum/aws_signing_client v0.0.0-20200229211254-f7815c59d5c1
 	github.com/sirupsen/logrus v1.9.0
 	github.com/slack-go/slack v0.12.2
+	github.com/sourcegraph/conc v0.3.0
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spiffe/spire v1.5.6
@@ -221,7 +222,6 @@ require (
 	github.com/segmentio/backo-go v0.0.0-20200129164019-23eae7c10bd3 // indirect
 	github.com/sergi/go-diff v1.3.1 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
-	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
 	github.com/tinylib/msgp v1.1.8 // indirect
 	github.com/ulikunitz/xz v0.5.10 // indirect

go.sum

@@ -1248,12 +1248,10 @@ go.szostok.io/version v1.2.0 h1:8eMMdfsonjbibwZRLJ8TnrErY8bThFTQsZYV16mcXms=
 go.szostok.io/version v1.2.0/go.mod h1:EiU0gPxaXb6MZ+apSN0WgDO6F4JXyC99k9PIXf2k2E8=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
-go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
-go.uber.org/multierr v1.8.0 h1:dg6GjLku4EH+249NNmoIciG9N/jURbDG+pFlTkhzIC8=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=

pkg/bot/discord.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/bwmarrin/discordgo"
 	"github.com/sirupsen/logrus"
+	"github.com/sourcegraph/conc/pool"
 
 	"github.com/kubeshop/botkube/pkg/api"
 	"github.com/kubeshop/botkube/pkg/bot/interactive"
@@ -39,17 +40,20 @@ const (
 
 // Discord listens for user's message, execute commands and sends back the response.
 type Discord struct {
-	log             logrus.FieldLogger
-	executorFactory ExecutorFactory
-	reporter        AnalyticsReporter
-	api             *discordgo.Session
-	botID           string
-	channelsMutex   sync.RWMutex
-	channels        map[string]channelConfigByID
-	notifyMutex     sync.Mutex
-	botMentionRegex *regexp.Regexp
-	commGroupName   string
-	renderer        *DiscordRenderer
+	log                   logrus.FieldLogger
+	executorFactory       ExecutorFactory
+	reporter              AnalyticsReporter
+	api                   *discordgo.Session
+	botID                 string
+	channelsMutex         sync.RWMutex
+	channels              map[string]channelConfigByID
+	notifyMutex           sync.Mutex
+	botMentionRegex       *regexp.Regexp
+	commGroupName         string
+	renderer              *DiscordRenderer
+	messages              chan discordMessage
+	discordMessageWorkers *pool.Pool
+	shutdownOnce          sync.Once
 }
 
 // discordMessage contains message details to execute command and send back the result.
@@ -75,19 +79,33 @@ func NewDiscord(log logrus.FieldLogger, commGroupName string, cfg config.Discord
 	}
 
 	return &Discord{
-		log:             log,
-		reporter:        reporter,
-		executorFactory: executorFactory,
-		api:             api,
-		botID:           cfg.BotID,
-		commGroupName:   commGroupName,
-		channels:        channelsCfg,
-		botMentionRegex: botMentionRegex,
-		renderer:        NewDiscordRenderer(),
+		log:                   log,
+		reporter:              reporter,
+		executorFactory:       executorFactory,
+		api:                   api,
+		botID:                 cfg.BotID,
+		commGroupName:         commGroupName,
+		channels:              channelsCfg,
+		botMentionRegex:       botMentionRegex,
+		renderer:              NewDiscordRenderer(),
+		messages:              make(chan discordMessage, 100),
+		discordMessageWorkers: pool.New().WithMaxGoroutines(10),
 	}, nil
 }
 
-// Start starts the Discord websocket connection and listens for messages.
+func (b *Discord) startMessageProcessor(ctx context.Context) {
+	b.log.Info("Starting discord message processor...")
+	defer b.log.Info("Stopped discord message processor...")
+
+	for msg := range b.messages {
+		b.discordMessageWorkers.Go(func() {
+			err := b.handleMessage(ctx, msg)
+			if err != nil {
+				b.log.Errorf("while handling message: %s", err.Error())
+			}
+		})
+	}
+} // Start starts the Discord websocket connection and listens for messages.
 func (b *Discord) Start(ctx context.Context) error {
 	b.log.Info("Starting bot")
 
@@ -113,14 +131,10 @@ func (b *Discord) Start(ctx context.Context) error {
 	}
 
 	b.log.Info("Botkube connected to Discord!")
-
+	go b.startMessageProcessor(ctx)
 	<-ctx.Done()
 	b.log.Info("Shutdown requested. Finishing...")
-	err = b.api.Close()
-	if err != nil {
-		return fmt.Errorf("while closing connection: %w", err)
-	}
-
+	b.shutdown()
 	return nil
 }
 
@@ -345,6 +359,19 @@ func (b *Discord) formatMessage(msg interactive.CoreMessage) (*discordgo.Message
 	}, nil
 }
 
+func (b *Discord) shutdown() {
+	b.shutdownOnce.Do(func() {
+		b.log.Info("Shutting down discord message processor...")
+		close(b.messages)
+		b.discordMessageWorkers.Wait()
+
+		err := b.api.Close()
+		if err != nil {
+			b.log.Errorf("While closing connection: %v", err)
+		}
+	})
+}
+
 func discordChannelsConfigFrom(log logrus.FieldLogger, api *discordgo.Session, channelsCfg config.IdentifiableMap[config.ChannelBindingsByID]) (map[string]channelConfigByID, error) {
 	res := make(map[string]channelConfigByID)
 	for channAlias, channCfg := range channelsCfg {

pkg/bot/mattermost.go

@@ -5,14 +5,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/sourcegraph/conc/pool"
 	"net/url"
 	"regexp"
 	"strings"
 	"sync"
 
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/sirupsen/logrus"
+	"github.com/sourcegraph/conc/pool"
 
 	"github.com/kubeshop/botkube/pkg/api"
 	"github.com/kubeshop/botkube/pkg/bot/interactive"
@@ -176,6 +176,9 @@ func (b *Mattermost) Start(ctx context.Context) error {
 	// For now, we are adding retry logic to reconnect to the server
 	// https://github.com/kubeshop/botkube/issues/201
 	b.log.Info("Botkube connected to Mattermost!")
+
+	go b.startMessageProcessor(ctx)
+
 	for {
 		select {
 		case <-ctx.Done():