Parallel message processing.

CHANGELOG.md

@@ -1,3 +1,6 @@
+2023-11-07 v1.6.0
+  - Parallel message processing.
+
 2023-08-31 v1.5.6
   - Ban old users with first message (bigger newbie store period).
 

config/cnf.go

@@ -10,11 +10,12 @@ type Cnf struct {
 }
 
 type Tg struct {
-	BotName string
-	Path    string
-	Proxy   string
-	Token   string
-	URL     string
+	BotName    string
+	Path       string
+	Processors int `default:"10"`
+	Proxy      string
+	Token      string
+	URL        string
 }
 
 type DB struct {

telegram/telegram.go

@@ -28,6 +28,7 @@ type InstanceObj struct {
 	router *chi.Mux
 	stop   chan bool
 	stor   *storage.InstanceObj
+	upd    tgbotapi.UpdatesChannel
 }
 
 type Options struct {
@@ -80,7 +81,7 @@ func (hdl *InstanceObj) Start() error {
 
 	hdl.oldLog.Info(resp.Description)
 
-	upd := hdl.bot.ListenForWebhook("/" + hdl.cnf.Path)
+	hdl.upd = hdl.bot.ListenForWebhook("/" + hdl.cnf.Path)
 
 	// HACK TODO
 	// We must register our handler again in internal router
@@ -90,6 +91,8 @@ func (hdl *InstanceObj) Start() error {
 		r.Handle("/"+hdl.cnf.Path, http.DefaultServeMux)
 	})
 
+	hdl.processors()
+
 	hdl.lock.Add(1)
 	defer hdl.lock.Done()
 
@@ -106,16 +109,32 @@ func (hdl *InstanceObj) Start() error {
 		case <-hdl.stop:
 			tick.Stop()
 			return nil
-		case msg := <-upd:
-			ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+		}
+	}
+}
 
-			err := hdl.process(ctx, msg)
-			if err != nil {
-				hdl.oldLog.Error(err)
+func (hdl *InstanceObj) processors() {
+	hdl.lock.Add(hdl.cnf.Processors)
+
+	for i := 0; i < hdl.cnf.Processors; i++ {
+		go func() {
+			for {
+				select {
+				case <-hdl.stop:
+					hdl.lock.Done()
+					return
+				case msg := <-hdl.upd:
+					ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+
+					err := hdl.process(ctx, msg)
+					if err != nil {
+						hdl.oldLog.Error(err)
+					}
+
+					cancel()
+				}
 			}
-
-			cancel()
-		}
+		}()
 	}
 }