package nsq

import (
	"apsClient/pkg/logx"
	"encoding/json"
	"fmt"
	"time"
)

type Caller interface {
	Call(msg []byte, duration time.Duration) ([]byte, error)
}

type DefaultCaller struct {
	NsqChannel    string
	RequestTopic  string
	ResponseTopic string
}

func NewCaller(requestTopic, responseTopic string) *DefaultCaller {
	return &DefaultCaller{
		NsqChannel:    "",
		RequestTopic:  requestTopic,
		ResponseTopic: responseTopic,
	}
}

func (caller *DefaultCaller) Call(input interface{}, output interface{}, timeout time.Duration) error {
	msg, err := json.Marshal(input)
	if err != nil {
		return err
	}
	producer := GetProducer()
	err = producer.Publish(caller.RequestTopic, msg)
	if err != nil {
		return err
	}
	to := time.After(timeout)
	for {
		select {
		case <-to:
			logx.Errorf("message call failed due to timeout, request topic: %v, msg: %+v, responseTopic", caller.RequestTopic, input, caller.ResponseTopic)
			return fmt.Errorf("wait response message timeout when request: %v", caller.ResponseTopic)
		case data := <-ReceivedMessageChan:
			if data.Topic == caller.ResponseTopic {
				return json.Unmarshal(data.Message, &output)
			} else {
				//把消息放回去
				ReceivedMessageChan <- data
			}
		}
	}
}

func (caller *DefaultCaller) Send(input interface{}) error {
	msg, err := json.Marshal(input)
	if err != nil {
		return err
	}
	producer := GetProducer()
	err = producer.Publish(caller.RequestTopic, msg)
	return err

}