package work

import (
	"analysis/logo"
	"context"

	"time"

	"basic.com/valib/deliver.git"

	"basic.com/pubsub/protomsg.git"
	"github.com/gogo/protobuf/proto"
)

// Reciever recv from ipc
type Reciever struct {
	ctx    context.Context
	ipcURL string
	chMsg  chan MsgRS

	shm bool
}

// NewReciever new recv
func NewReciever(url string, chMsg chan MsgRS, shm bool) *Reciever {
	return &Reciever{
		ipcURL: url,
		chMsg:  chMsg,
		shm:    shm,
	}
}

func (r *Reciever) unserilizeProto(ctx context.Context, data <-chan []byte) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			d := <-data
			if len(d) < 100 {
				continue
			}
			// logo.Infoln(len(d), "reciver数据")
			msg := protomsg.SdkMessage{}
			if err := proto.Unmarshal(d, &msg); err != nil {
				logo.Errorln(err, " msg 处理异常")
				continue
			}
			// logo.Infoln("RECV MSG: ", msg.Cid, " TASK: ", msg.Tasklab.Taskid, " SDK: ", msg.Tasklab.Sdkinfos[msg.Tasklab.Index].Sdktype)
			outMsg := MsgRS{Msg: msg}
			r.chMsg <- outMsg
		}
	}
}

func (r *Reciever) run(ctx context.Context, i deliver.Deliver) {
	dataChan := make(chan []byte)

	go r.unserilizeProto(ctx, dataChan)

	// t := time.Now()
	// sc := 0

	for {
		select {
		case <-ctx.Done():
			i.Close()
			return
		default:

			if r.shm {
				if d, err := i.Recv(); err != nil {
					i.Close()
					logo.Infoln("ANALYSIS RECV ERROR: ", err)

					c, err := deliver.NewClientWithError(deliver.Shm, r.ipcURL)
					for {
						if err == nil {
							break
						}
						time.Sleep(time.Second)
						c, err = deliver.NewClientWithError(deliver.Shm, r.ipcURL)
						logo.Infoln("ANALYSIS CREATE FAILED : ", err)
					}
					i = c
					logo.Infoln("ANALYSIS CREATE SHM")
				} else {
					if d != nil {
						logo.Infoln("~~~~~~shm recv image:", len(d))
						dataChan <- d
					}
				}
			} else {
				if msg, err := i.Recv(); err != nil {
					// logo.Errorln("recv error : ", err, " url: ", r.ipcURL)
				} else {
					logo.Infoln("~~~~~~mangos recv image:", len(msg))
					dataChan <- msg
				}
			}

			// sc++
			// if sc == 25 {
			// 	logo.Infoln("SDK RECV 25 FRAME USE TIME: ", time.Since(t))
			// 	sc = 0
			// 	t = time.Now()
			// }

		}
	}
}

// Run run a IPC client
func (r *Reciever) Run(ctx context.Context) {
	if r.shm {
		r.runShm(ctx)
	} else {
		r.run(ctx, deliver.NewClient(mode, r.ipcURL))
	}
}

func (r *Reciever) runShm(ctx context.Context) {
	c, err := deliver.NewClientWithError(deliver.Shm, r.ipcURL)
	for {
		if err == nil {
			break
		}
		time.Sleep(1 * time.Second)
		c, err = deliver.NewClientWithError(deliver.Shm, r.ipcURL)
		logo.Infoln("CLIENT CREATE FAILED : ", err)
	}
	r.run(ctx, c)
}

///////////////////////////////////////////////////

///////// test

///////////////////////////////////////////////////

// Run2 run a IPC server or client
func (r *Reciever) Run2(ctx context.Context, server bool) {
	if server {
		r.run(ctx, deliver.NewServer(deliver.PushPull, r.ipcURL))
	} else {
		r.run(ctx, deliver.NewClient(deliver.PushPull, r.ipcURL))
	}
}