package service

import (
	"context"
	"fmt"
	"os"
	"strings"
	"time"

	"nanomsg.org/go-mangos"
	"nanomsg.org/go-mangos/protocol/pull"
	"nanomsg.org/go-mangos/transport/ipc"
	"nanomsg.org/go-mangos/transport/tcp"
)

// 山东断流监控
func rmExistedIpcName(url string) {
	s := strings.Split(url, "://")

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

func newPull(ctx context.Context, url string, timeout int) mangos.Socket {
	rmExistedIpcName(url)

	sock, err := pull.NewSocket()
loop1:
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			if err == nil {
				break loop1
			}
			time.Sleep(time.Second)
			rmExistedIpcName(url)
			sock, err = pull.NewSocket()
			fmt.Println("!!!!!!Pull can't get new socket:", err)
		}
	}

	sock.AddTransport(ipc.NewTransport())
	sock.AddTransport(tcp.NewTransport())
	sock.SetOption(mangos.OptionRecvDeadline, time.Duration(timeout)*time.Second)
	sock.SetOption(mangos.OptionSendDeadline, time.Duration(timeout)*time.Second)

	err = sock.Listen(url)
loop2:
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			if err == nil {
				break loop2
			}
			time.Sleep(time.Second)
			rmExistedIpcName(url)
			err = sock.Listen(url)

			fmt.Println("!!!!!!Pull can't listen socket:", err, "URL:", url)
		}
	}

	return sock
}