new-server/app/services/publish.go

package services

import (
	"context"
	"git.aiterp.net/lucifer/new-server/app/config"
	"git.aiterp.net/lucifer/new-server/models"
	"log"
	"sync"
	"time"
)

func StartPublisher() {
	ctx := context.Background()

	go func() {
		for devices := range config.PublishChannel {
			if len(devices) == 0 {
				continue
			}

			lists := make(map[int][]models.Device, 4)
			for _, device := range devices {
				lists[device.BridgeID] = append(lists[device.BridgeID], device)
			}

			ctx, cancel := context.WithTimeout(ctx, time.Second * 30)

			bridges, err := config.BridgeRepository().FetchAll(ctx)
			if err != nil {
				log.Println("Publishing error (1): " + err.Error())
				cancel()
				continue
			}

			wg := sync.WaitGroup{}
			for _, devices := range lists {
				wg.Add(1)

				go func(devices []models.Device) {
					defer wg.Done()

					var bridge models.Bridge
					for _, bridge2 := range bridges {
						if bridge2.ID == devices[0].BridgeID {
							bridge = bridge2
						}
					}
					if bridge.ID == 0 {
						log.Println("Unknown bridge")
						return
					}

					if bridge.Driver == models.DTLIFX {
						return
					}

					bridge, err := config.BridgeRepository().Find(ctx, devices[0].BridgeID)
					if err != nil {
						log.Println("Publishing error (1): " + err.Error())
						return
					}

					driver, err := config.DriverProvider().Provide(bridge.Driver)
					if err != nil {
						log.Println("Publishing error (2): " + err.Error())
						return
					}

					err = driver.Publish(ctx, bridge, devices)
					if err != nil {
						log.Println("Publishing error (3): " + err.Error())
						return
					}
				}(devices)
			}

			wg.Wait()
			cancel()
		}
	}()
}