update change pool address send authorize info

internal/miner/miner.go

@@ -35,8 +35,8 @@ func NewMiner(coin string, poolAddress string, minerConn net.Conn) (*Miner, erro
 	}, nil
 }
 
-// 动态切换矿池地址并建立新连接
-func (m *Miner) ChangePoolAddress(newAddress string) {
+// 动态切换矿池地址并建立新连接,切换时由proxy构造，由于每个池子的authorize认证不同，需单独处理
+func (m *Miner) ChangePoolAddress(newAddress string, authorizeMsg string) {
 	m.Lock()
 	defer m.Unlock()
 
@@ -54,7 +54,7 @@ func (m *Miner) ChangePoolAddress(newAddress string) {
 
 	m.PoolConn = newConn
 	m.PoolAddress = newAddress
-
+	m.PoolConn.Write([]byte(authorizeMsg + "\n")) // 发送认证信息
 	fmt.Println("成功切换矿池地址为:", newAddress)
 }
 
@@ -78,10 +78,10 @@ func (m *Miner) HandleMinerMsg(ch chan string) {
 		case "nexa":
 			//
 		default:
-			var msg msg.Authorize_msg
-			if err := json.Unmarshal([]byte(msgStr), &msg); err == nil &&
-				msg.Method == "mining.authorize" && len(msg.Params) >= 1 && !sent {
-				parts := strings.Split(msg.Params[0], ".")
+			var _msg msg.Authorize_msg
+			if err := json.Unmarshal([]byte(msgStr), &_msg); err == nil &&
+				_msg.Method == "mining.authorize" && len(_msg.Params) >= 1 && !sent {
+				parts := strings.Split(_msg.Params[0], ".")
 				if len(parts) >= 2 {
 					userSign := parts[0] + "-" + parts[1]
 					select {

internal/msg/msg.go

@@ -4,6 +4,7 @@ type ZmqMsg struct {
 	MethodID int    `json:"methodId"` //0(add), 1(delete)
 	ID       string `json:"id"`       // user-miner
 	Address  string `json:"address"`  // 转发目标地址
+	Msg      string `json:"msg"`      // 转发矿池的mining.authorize方法配置参数
 }
 
 type Authorize_msg struct {

internal/proxy.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"os"
 	"proxy/internal/miner"
+	"proxy/internal/msg"
 	"proxy/internal/zmq"
 	"sync"
 	"time"
@@ -18,6 +19,7 @@ type Config struct {
 	ZmqAddr     string `json:"zmqAddr"`
 	TcpAddr     string `json:"tcpAddr"`
 	DefaultAddr string `json:"proxyAddr"`
+	RabbitTopic string `json:"rabbitTopic"`
 }
 
 type ProxyCtx struct {
@@ -26,7 +28,7 @@ type ProxyCtx struct {
 	Cfg       Config
 	Listener  net.Listener
 	MinerConn []*miner.Miner
-	ProxyList map[string]string
+	ProxyList map[string]msg.ZmqMsg
 }
 
 func initConfig() Config {
@@ -52,7 +54,7 @@ func newProxy() *ProxyCtx {
 		Cfg:       cfg,
 		Listener:  listener,
 		MinerConn: []*miner.Miner{},
-		ProxyList: make(map[string]string),
+		ProxyList: make(map[string]msg.ZmqMsg),
 	}
 }
 
@@ -79,10 +81,12 @@ func (p *ProxyCtx) handleMinerConnect() {
 			select {
 			case userSign := <-ch:
 				p.mu.RLock()
-				addr, ok := p.ProxyList[userSign]
+				zmqMsg, ok := p.ProxyList[userSign]
 				p.mu.RUnlock()
 				if ok {
-					minerObj.ChangePoolAddress(addr)
+					addr := zmqMsg.Address
+					authorizeMsg := zmqMsg.Msg
+					minerObj.ChangePoolAddress(addr, authorizeMsg)
 				}
 
 				p.mu.Lock()
@@ -101,7 +105,7 @@ func (p *ProxyCtx) handleMinerConnect() {
 }
 
 func (p *ProxyCtx) startZMQ() {
-	zmq.StartZMQ(p.Cfg.ZmqAddr, "testQueue", p.ProxyList, &p.mu)
+	zmq.StartZMQ(p.Cfg.ZmqAddr, p.Cfg.RabbitTopic, p.ProxyList, &p.mu)
 }
 
 func StartProxy() {

internal/zmq/zmq.go

@@ -101,7 +101,7 @@ func initRabbitConsumer(rabbitURL, queueName string) (*amqp.Connection, *amqp.Ch
 	return conn, ch, msgs, nil
 }
 
-func handleRabbitMsg(body []byte, proxyList map[string]string, proxyListLock *sync.RWMutex) {
+func handleRabbitMsg(body []byte, proxyList map[string]msg.ZmqMsg, proxyListLock *sync.RWMutex) {
 	var data msg.ZmqMsg
 	if err := json.Unmarshal(body, &data); err != nil {
 		fmt.Println("[rabbitmq]: json unmarshal error:", err)
@@ -110,13 +110,13 @@ func handleRabbitMsg(body []byte, proxyList map[string]string, proxyListLock *sy
 	proxyListLock.Lock()
 	defer proxyListLock.Unlock()
 	if data.MethodID == 0 {
-		proxyList[data.ID] = data.Address
+		proxyList[data.ID] = data
 	} else {
 		delete(proxyList, data.ID)
 	}
 }
 
-func StartZMQ(rabbitURL, queueName string, proxyList map[string]string, proxyListLock *sync.RWMutex) {
+func StartZMQ(rabbitURL, queueName string, proxyList map[string]msg.ZmqMsg, proxyListLock *sync.RWMutex) {
 	conn, ch, msgs, err := initRabbitConsumer(rabbitURL, queueName)
 	if err != nil {
 		fmt.Println("[rabbitmq]:", err)