package main

import (
	"fmt"
	"gitee.com/liujinsuo/tool/toolqueue"
	"github.com/pkg/errors"
	"log"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"runtime"
	"sync/atomic"
	"syscall"
	"time"
)

// 实测性能5秒消费1000万数据
func main() {
	log.SetFlags(log.Lshortfile | log.LstdFlags)

	go func() {
		log.Println(http.ListenAndServe("localhost:6060", nil))
	}()

	sub := atomic.Int64{}
	queue := toolqueue.NewQueue[int](10000000, func(QueueName toolqueue.QueueName, dataList []int) {
		//log.Printf("消费数据 name=%v %+v", QueueName, dataList)
		//time.Sleep(time.Second * 1)
		sub.Add(int64(len(dataList)))
		if sub.Load()%10000000 == 0 {
			log.Printf("消费到%v", sub.Load())
		}
	})

	queue.SetLogCb(func(err error) {
		log.Printf("err=%s", err)
	}, func(msg string) {
		log.Printf("msg=%s", msg)
	}, func(msg string) {
		log.Printf("debug=%s", msg)
	})

	add := atomic.Int64{}
	go func() {
		log.Printf("开始添加数据")
		for i := 0; i < 10; i++ {

			go func(i int) {
				for j := 0; j < 20000000; j++ {
					que := toolqueue.QueueName(fmt.Sprintf("queue_name%d", i))
					//log.Printf("添加数据 que=%v data=%v", que, j)
					if err := queue.Publish(que, j); errors.Is(err, toolqueue.ErrQueueFull) {
						log.Printf("队列已满 que=%v %v", que, err)
						time.Sleep(time.Millisecond * 100)
						j--
						continue
					} else if err != nil {
						log.Printf("que=%v %v", que, err)
						break
					}
					add.Add(1)
					runtime.Gosched()
					//log.Printf("添加数据 que=%v data=%v", que, j)
					//time.Sleep(time.Millisecond * 6000)
				}

			}(i)
		}
		log.Printf("数据添加完成")
	}()

	queue.Start()

	quit := make(chan os.Signal)
	signal.Notify(quit, syscall.SIGHUP, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM)
	sign := <-quit //阻塞等待结束信号
	log.Println(sign, "----------")
	log.Printf("add %v sub %v", add.Load(), sub.Load())

	queue.Stop()
	log.Println(sign, "----------")
	queue.Wait()
	log.Println(sign, "----------")

}