watcher/pkg/pools/proxy.go

package pools

import (
	"context"
	"fmt"
	"github.com/golang/glog"
	"github.com/timerzz/proxypool/pkg/getter"
	"github.com/timerzz/proxypool/pkg/proxy"
	"github.com/timerzz/proxypool/pkg/tool"
	"log/slog"
	"math/rand"
	"sync"
	"time"
)

type ProxyPool struct {
	m          sync.Mutex
	proxies    proxy.ProxyList
	subscribes []string //订阅url
	updated    time.Time
}

func NewProxyPool(subscribes []string) *ProxyPool {
	var p = &ProxyPool{}
	p.subscribes = subscribes
	p.Update()
	return p
}

func (p *ProxyPool) Status() (proxy.ProxyList, time.Time) {
	return p.proxies, p.updated
}

// Update 更新代理池
func (p *ProxyPool) Update() {
	var list = make(proxy.ProxyList, 0, len(p.proxies))
	for _, url := range p.subscribes {
		subscribeGetter, err := getter.NewSubscribe(tool.Options{"url": url})
		if err != nil {
			slog.Warn(fmt.Sprintf("创建Subscribe Getter失败：%v", err))
			continue
		}
		list = list.UniqAppendProxyList(subscribeGetter.Get())
	}
	glog.Infof("代理源共 %d 个: %v", len(p.subscribes), p.subscribes)
	glog.Infof("获取代理共 %d 个", len(list))
	p.m.Lock()
	p.proxies = list
	p.m.Unlock()
	p.updated = time.Now()
}

// CronUpdate 定时更新
func (p *ProxyPool) CronUpdate(ctx context.Context, interval time.Duration) {
	if interval == 0 {
		interval = time.Minute * 30
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			p.Update()
		}
	}
}

// RandomIterator 获取随机代理的迭代器
func (p *ProxyPool) RandomIterator() func() proxy.Proxy {
	return func() (proxy proxy.Proxy) {
		if len(p.proxies) == 0 {
			return nil
		}
		p.m.Lock()
		defer p.m.Unlock()
		curIndex := rand.Intn(len(p.proxies))
		return p.proxies[curIndex]
	}
}