package profile

import (
	"context"
	"demo/deliver"
	"fmt"
	"os"
	"sync"
	"time"
)

func shmrecver(ctx context.Context, c deliver.Deliver, index int, ch chan<- bool) {

	var msg []byte
	var err error

	var t int64
	var elapse int64
	count := 0

	for {
		select {
		case <-ctx.Done():
			return
		default:
			msg, err = c.Recv()
			if err != nil {
				c.Close()
				url := "hello"
				i, err := deliver.NewClientWithError(deliver.Shm, url)
				for {
					if err == nil {
						break
					}
					time.Sleep(1 * time.Second)
					i, err = deliver.NewClientWithError(deliver.Shm, url)

					fmt.Println("client create failed : ", err)

				}
				c = i

				fmt.Println("recv error : ", err)
				continue
			}
			if ch != nil {
				ch <- true
			}

			if t == 0 {
				t = time.Now().UnixNano()
			}
			elapse = time.Now().UnixNano() - t

			count++

			if elapse > 1e9 {
				fmt.Printf("NODE-%d: RECEIVED \"%d\" data len %d, use \"%d\" ns\n",
					index, count, len(msg), elapse)
				elapse = 0
				count = 0
				t = 0
			}
		}

		// time.Sleep(10 * time.Millisecond)
	}

}

func shmrecver2(ctx context.Context, c deliver.Deliver, index int, ch chan<- bool, pool *sync.Pool) {

	// var msg []byte
	// var err error

	var t int64
	var elapse int64
	count := 0

	for {
		select {
		case <-ctx.Done():
			return
		default:
			msg := *pool.Get().(*[]byte)
			_, err := c.Recv2(msg)
			if err != nil {
				c.Close()
				url := "hello"
				i, err := deliver.NewClientWithError(deliver.Shm, url)
				for {
					if err == nil {
						break
					}
					time.Sleep(1 * time.Second)
					i, err = deliver.NewClientWithError(deliver.Shm, url)

					fmt.Println("client create failed : ", err)

				}
				c = i

				pool.Put(&msg)

				fmt.Println("recv error : ", err)
				continue
			}
			if ch != nil {
				ch <- true
			}

			if t == 0 {
				t = time.Now().UnixNano()
			}
			elapse = time.Now().UnixNano() - t

			count++

			if elapse > 1e9 {
				fmt.Printf("NODE-%d: RECEIVED \"%d\" data len %d, use \"%d\" ns\n",
					index, count, len(msg), elapse)
				elapse = 0
				count = 0
				t = 0
			}

			pool.Put(&msg)
		}

		// time.Sleep(10 * time.Millisecond)
	}

}

var chunkSize = 32 * 1024 * 1024

func Shmrecv(ctx context.Context, server bool, ipc string, count int) {
	blocks := 2

	if server {
		s, err := deliver.NewServerWithError(deliver.Shm, ipc, blocks, dLen)
		for {
			if err == nil {
				break
			}
			time.Sleep(1 * time.Second)
			s, err = deliver.NewServerWithError(deliver.Shm, ipc, blocks, dLen)
		}
		go shmrecver(ctx, s, 0, nil)

	} else {
		recvers(ctx, ipc, count, nil)
	}
}

func recvers(ctx context.Context, url string, count int, ch chan<- bool) []deliver.Deliver {
	for {
		file := "/dev/shm/" + url
		exist, _ := pathExists(file)
		if exist {
			break
		}
		fmt.Println("wait for shm server start")
		time.Sleep(time.Second)
	}

	// pool := &sync.Pool{
	// 	New: func() interface{} {
	// 		b := make([]byte, chunkSize)
	// 		return &b
	// 	},
	// }

	// pool := slab.NewAtomPool(chunkSize, chunkSize, 2, chunkSize*8)

	var cs []deliver.Deliver
	for i := 0; i < count; i++ {
		c, err := deliver.NewClientWithError(deliver.Shm, url)
		for {
			if err == nil {
				break
			}
			time.Sleep(1 * time.Second)
			c, err = deliver.NewClientWithError(deliver.Shm, url)
			fmt.Println(i, " client create failed : ", err)
		}
		cs = append(cs, c)
		go shmrecver(ctx, c, i, ch)
		// go shmrecver2(ctx, c, i, ch, pool)
	}
	return cs
}

func pathExists(path string) (bool, error) {
	_, err := os.Stat(path)
	if err == nil {
		return true, nil
	}
	if os.IsNotExist(err) {
		return false, nil
	}
	return false, err
}