package deliver

import (
	"errors"
	"fmt"
	"os"
	"strings"
	"time"

	"nanomsg.org/go-mangos"
	"nanomsg.org/go-mangos/protocol/bus"
	"nanomsg.org/go-mangos/protocol/pair"
	"nanomsg.org/go-mangos/protocol/pub"
	"nanomsg.org/go-mangos/protocol/pull"
	"nanomsg.org/go-mangos/protocol/push"
	"nanomsg.org/go-mangos/protocol/rep"
	"nanomsg.org/go-mangos/protocol/req"
	"nanomsg.org/go-mangos/protocol/respondent"
	"nanomsg.org/go-mangos/protocol/sub"
	"nanomsg.org/go-mangos/protocol/surveyor"
	"nanomsg.org/go-mangos/transport/all"
)

// NNG mangos wrap
type NNG struct {
	sock mangos.Socket
	raw  bool
}

// Send impl interface Diliver
func (n *NNG) Send(data []byte) error {
	if n.sock == nil {
		return errors.New("please init NNG first")
	}

	if surveyorTime > 0 {
		time.Sleep(time.Duration(surveyorTime*2) * time.Second)
	}

	if n.raw {
		msg := mangos.NewMessage(len(data))
		msg.Body = data
		return n.sock.SendMsg(msg)
	}

	return n.sock.Send(data)
}

// Recv impl interface Diliver
func (n *NNG) Recv() ([]byte, error) {
	if n.sock == nil {
		return nil, errors.New("please init NNG first")
	}
	if n.raw {
		var msg *mangos.Message
		var err error
		if msg, err = n.sock.RecvMsg(); err != nil {
			return nil, err
		}
		return msg.Body, nil
	}
	return n.sock.Recv()
}

// Close impl interface Deliver
func (n *NNG) Close() {
	if n.sock != nil {
		n.sock.Close()
		n.sock = nil
	}
}

func nngPush(m Mode, url string, args ...interface{}) *NNG {

	rmExistedIpcName(url)
	if sock, err := newSocket(protoProducer(m)); err == nil {
		if err = setSocketOptions(sock, args); err != nil {
			sock.Close()
			sock = nil
			return nil
		}
		if err = sock.Listen(url); err != nil {
			sock.Close()
			sock = nil
			return nil
		}
		return &NNG{
			sock,
			true,
		}
	}

	return nil
}

func nngPull(m Mode, url string, args ...interface{}) *NNG {

	if sock, err := newSocket(protoConsumer(m)); err == nil {
		if err = setSocketOptions(sock, args); err != nil {
			sock.Close()
			sock = nil
			return nil
		}

		if err = sock.Dial(url); err != nil {
			sock.Close()
			sock = nil
			return nil
		}

		return &NNG{
			sock,
			true,
		}
	}

	return nil
}

// maxRecvSize max recv size
var (
	maxRecvSize  = 33 * 1024 * 1024
	surveyorTime = -1
)

func defualtSocketOptions() map[string]interface{} {

	options := make(map[string]interface{})

	options[mangos.OptionMaxRecvSize] = maxRecvSize
	options[mangos.OptionWriteQLen] = 0
	options[mangos.OptionReadQLen] = 0
	options[mangos.OptionRecvDeadline] = time.Second
	options[mangos.OptionSendDeadline] = time.Second
	options[mangos.OptionRaw] = true

	return options
}

func setSocketOptions(sock mangos.Socket, args ...interface{}) error {

	options := defualtSocketOptions()

	switch sock.GetProtocol().Number() {
	case mangos.ProtoSub:
		for _, arg := range args {
			switch arg.(type) {
			case string:
				options[mangos.OptionSubscribe] = []byte(arg.(string))
			default:
				options[mangos.OptionSubscribe] = []byte("")
			}
		}
	case mangos.ProtoSurveyor:
		for _, arg := range args {
			switch arg.(type) {
			case int:
				if arg.(int) < 2 {
					surveyorTime = 1
				} else {
					surveyorTime = arg.(int) / 2
				}
			default:
			}
			options[mangos.OptionSurveyTime] = time.Duration(surveyorTime) * time.Second
		}
	default:
		fmt.Println("no additional args")
	}

	for k, v := range options {
		if err := sock.SetOption(k, v); err != nil {
			return err
		}
	}
	return nil
}

func rmExistedIpcName(url string) {
	s := strings.Split(url, "://")

	if s[0] == "ipc" {
		if _, err := os.Stat(s[1]); err == nil {
			os.Remove(s[1])
		}
	}
}

// newSocket allocates a new Socket.  The Socket is the handle used to
// access the underlying library.
func newSocket(p protocol) (mangos.Socket, error) {

	var s mangos.Socket
	var err error

	switch p {
	case PUB:
		s, err = pub.NewSocket()
	case SUB:
		s, err = sub.NewSocket()
	case PUSH:
		s, err = push.NewSocket()
	case PULL:
		s, err = pull.NewSocket()
	case REQ:
		s, err = req.NewSocket()
	case REP:
		s, err = rep.NewSocket()
	case SURVEYOR:
		s, err = surveyor.NewSocket()
	case RESPONDENT:
		s, err = respondent.NewSocket()
	case PAIR:
		s, err = pair.NewSocket()
	case BUS:
		s, err = bus.NewSocket()
	default:
		err = mangos.ErrBadProto
	}

	if err != nil {
		return nil, err
	}

	all.AddTransports(s)

	return s, nil
}

func die(format string, v ...interface{}) {
	fmt.Fprintln(os.Stderr, fmt.Sprintf(format, v...))
	os.Exit(1)
}

// Protocol is the numeric abstraction to the various protocols or patterns
// that Mangos supports.
type protocol int

// Constants for protocols.
const (
	PUSH       = protocol(mangos.ProtoPush)
	PULL       = protocol(mangos.ProtoPull)
	PUB        = protocol(mangos.ProtoPub)
	SUB        = protocol(mangos.ProtoSub)
	REQ        = protocol(mangos.ProtoReq)
	REP        = protocol(mangos.ProtoRep)
	SURVEYOR   = protocol(mangos.ProtoSurveyor)
	RESPONDENT = protocol(mangos.ProtoRespondent)
	BUS        = protocol(mangos.ProtoBus)
	PAIR       = protocol(mangos.ProtoPair)
)

func protoProducer(m Mode) protocol {
	switch m {
	case PushPull:
		return PUSH
	case PubSub:
		return PUB
	case ReqRep:
		return REP
	case SurvResp:
		return SURVEYOR
	case Bus:
		return BUS
	case Pair:
		return PAIR
	}
	return PUSH
}

func protoConsumer(m Mode) protocol {
	switch m {
	case PushPull:
		return PULL
	case PubSub:
		return SUB
	case ReqRep:
		return REQ
	case SurvResp:
		return RESPONDENT
	case Bus:
		return BUS
	case Pair:
		return PAIR
	}
	return PULL
}