package test
import (
	"bytes"
	"fmt"
	"os/exec"
	"testing"
)

// cmd.go  (写在一个文件中方法读代码)
// 待测试方法,
func ExecShell(s string) (string, error) {
	cmd := exec.Command("/bin/bash", "-c", s)
	var out bytes.Buffer
	cmd.Stdout = &out
	err := cmd.Run()
	if err != nil {
		return "", err
	}
	return out.String(), nil
}

func testExecShell(t *testing.T) {
	t.Log("testExecShell")
	_, err := ExecShell("ls")
	if err != nil {
		t.Error(err)
	}
}

func testFun1(t *testing.T)  {
	t.Log("testFun1")
}

func beforeInit()  {
	fmt.Println("beforeInit")
}

func afterClear()  {
	fmt.Println("afterClear")
}

func TestAll(t *testing.T)  {
	t.Run("testExec..", testExecShell)
	t.Run("testfun1", testFun1)
}

// 在测试中增加前置/后置函数调用的方法, 思路: 自定义测试的新入口,
// 1 自定义testFun1, testFun2..方法 , test的t是小写, 这样不会被默认调用
// 2 自定义Test*方法, Test的T 大写, 这样可以被go test 工具识别
// 3 在Test*方法中按顺序 编写 t.Run("fun_name", fun)
// 4 编写TestMain 方法 , 调用beforeInit, m.Run()(会调用其它Test*方法) afterClear
// 当前不写TestMain方法, 直接在 Test*方法中调用也一样
func TestMain(m *testing.M)  {
	fmt.Println("TestMain")
	beforeInit()
	m.Run()
	afterClear()
	//m.Run()
}

// limiter.go 

package limiter

import (
	"fmt"
	"go.uber.org/zap"
	"station.grab/src/common"
	"sync"
	"time"
)

const (
	// MinimaLimit is the minimal concurrency limit
	MinimaLimit = 5
)

// Job is an interface for add jobs.
type Job interface {
	Run() (resp string, err error)
}

// EasyLimiter object
type EasyLimiter struct {
	semp chan struct{} // 控制并发的chan

	wg sync.WaitGroup // waitGroup 用于等待协程执行完成, 并关闭通道\清理资源

	jobChan    chan Job         // Job 队列(实现接口即可, 解耦了任务的具体实现)
	resultChan chan interface{} // job执行结果队列
}

func NewEasyLimiter(taskCount, limit int) *EasyLimiter {
	if limit <= MinimaLimit {
		limit = MinimaLimit
	}

	c := &EasyLimiter{
		semp:       make(chan struct{}, limit),
		resultChan: make(chan interface{}, taskCount),
		jobChan:    make(chan Job, taskCount),
	}

	// 创建后马上就监听job队列
	// job队列中有数据且semp队列未满 (满了会阻塞,以此来实现并发控制), 则取出job对象, 交给单独协程处理
	go func() {
		for job := range c.jobChan {
			//c.semp <- struct{}{}

			select {
			case c.semp <- struct{}{}:

			case <-time.After(time.Millisecond * 200):
				common.ZLogger.Info("goroutine pool full, wait for 200 mis  ", zap.Int("size", len(c.semp)))
			}

			go func(ajob Job) {
				defer func() {
					c.wg.Done()
					<-c.semp
				}()
				//common.ZLogger.Info("开始执行任务")
				result, err := ajob.Run()
				//common.ZLogger.Info("完成执行任务")
				if err != nil {
					fmt.Printf("err:%v", err)
				}
				c.resultChan <- result

			}(job)
		}
		common.ZLogger.Info("task队列关闭")
	}()

	return c
}

func (c *EasyLimiter) AddJob(job Job) {
	c.wg.Add(1)
	c.jobChan <- job
	//common.ZLogger.Info("添加任务")
}

func (c *EasyLimiter) Wait() {
	// 关闭job队列 ,此时已不会再添加
	close(c.jobChan)
	c.wg.Wait()
	// 关闭result队列,以保证range方式读取chan 程序会正常向下执行
	close(c.resultChan)
}

// limiter_test.go 

package limiter

import (
	"fmt"
	"go.uber.org/zap"
	"math/rand"
	"station.grab/src/common"
	"testing"
	"time"
)

func Max(a int, b int) int { //注意参数和返回值是怎么声明的

	if a > b {
		return a
	}
	return b
}
func RandomInt(n int) int {
	rand.Seed(time.Now().UnixNano())
	v := rand.Intn(n)
	if v < 1 {
		v = 1
	}
	return v
}

type SampleJob struct {
	total int
	idx   int
	key   string
}

func (s SampleJob) Run() (resp string, err error) {

	v := fmt.Sprintf("job run: %d/%d, %v", s.idx, s.total, s.key)
	common.ZLogger.Info(v)
	//time.Sleep(time.Second*time.Duration(RandomInt(5)))
	time.Sleep(time.Second * 1)

	return v, nil
}

func TestLimiter_Execute(t *testing.T) {

	fmt.Println("begin")
	total := 20
	limiter := NewEasyLimiter(total, 5)

	for i := 0; i < total; i++ {
		limiter.AddJob(&SampleJob{
			total: total,
			idx:   i,
			key:   "test",
		})
	}

	// 控制完成并退出的信号
	chanSignal := make(chan interface{})

	go func() {
		for {
			select {
			case result, ok := <-limiter.resultChan:
				common.ZLogger.Info("read from result chan ", zap.Any("result", result))
				if !ok {
					common.ZLogger.Info("result通道关闭,退出当前goroutine")
					chanSignal<- 1
					return
				}
			}
		}
	}()

	limiter.Wait()

	<-chanSignal

	fmt.Println("done")
}

func TestRandom(t *testing.T) {
	for i := 0; i < 20; i++ {
		fmt.Println(RandomInt(5))
	}

}

type LineGraber struct {
	//最多允许n个并发同时执行
	ConcurrentNum int `json:"concurrent_num"`
}

func (grab LineGraber) Work() {

// "work start"

//最少5个协程一起工作
concurrentNum := grab.ConcurrentNum
if concurrentNum <= 0 {
    concurrentNum = 5
}
// sem通道容量限制了并发的协程数
sem := make(chan struct{}, concurrentNum)

// 准备待处理的任务列表  
taskLst := grab.buildTaskLst()

// 接口任务处理结果的通道
chanRespv := make(chan string, len(taskLst))

total := len(taskLst)
// log info fmt.Sprintf("tasksize:%d", total) 

var wg sync.WaitGroup
for idx, key := range taskLst {
    wg.Add(1)
    go func(taskv string, i int) {
   // 确保 wg.Done() (与add对应), 
   //     <-sem, 从semchan中取出(与下面 sem<- struct{}{}对应)
        defer func() {
            wg.Done()
            <-sem
        }()
        sem <- struct{}{}
        // 真实业务逻辑部分 
        // 方法中要将结果写入到chanRespv 
        // log info i, taskv 
        grab.processTask(taskv, oneproxy, chanRespv)

    }(key, idx)
}

wg.Wait()
close(chanRespv)

// range 可以从已经关闭的chan 中读取原有数据, 并退出  
for respv := range chanRespv {
    // log info read from chanResp
    var respResult RespResult
    err := json.Unmarshal([]byte(respv), &respResult)
    if err != nil {
        // log error decode异常 
        continue
    }
    // process ... 

}

// log info ("work done")

}

// 以上方法中在 wg.Wait之后才开始接收 chanRespv ,其实还可以优化\尽早提前处理chanRespv(可以及时读取到结果,避免所有任务完成后才读取到结果的情况), 只需新启协程在wg.Wait 之前接收处理 . 
// 要增加另外一个通道, 控制完成并退出的信号

// 在wg.wait 之前 
chanSignal := make(chan interface{})
go func() {
    for {
        select {
        case result, ok := <-limiter.resultChan:
            common.ZLogger.Info("done", zap.Any("result", result))
            if !ok {
                common.ZLogger.Info("result通道关闭,退出当前goroutine")
      // 通知主线程,可以继续向下执行了 
                chanSignal<- 1
                return
            }
        }
    }
}()

// 在程序最后添加 
<-chanSignal
	

// 业务逻辑处理  
// 注意chan 类型是只可写入 chan <-  string  
func (grab LineGraber) processTask(key string, proxy string, outchan chan<- string) {
	aimUrl := fmt.Sprintf(grap_base_url, key)
	respv, err := doget... 
	if err != nil {
		// log error ...
	}
	//log info ... 
	outchan <- respv
}