async_queue/util/publisher.go

package util

import (
	"app.yhyue.com/BP/queued/proto"
	"context"
	"google.golang.org/grpc"
	"log"
	"time"
)

/*
任务发布封装
*/
type Publisher struct {
	Id          string
	QueueServer string
	Channel     string
	C           <-chan *proto.PubReq
}

//
func NewPublisher(server string, channel string, ch <-chan *proto.PubReq) *Publisher {
	uuid, _ := makeUuid()
	return &Publisher{
		Id:          uuid,
		C:           ch,
		QueueServer: server,
		Channel:     channel,
	}
}

//
func (pb *Publisher) Run() {
	for {
		conn, err := grpc.Dial(pb.QueueServer, grpc.WithInsecure())
		if err != nil {
			time.Sleep(1 * time.Second)
			log.Fatalf("did not connect: %v", err)
		}
		c := proto.NewQueueServiceClient(conn)
		stream, _ := c.Publish(context.Background())
		//
	Lab:
		for {
			select {
			case msg := <-pb.C:
				uuid, _ := makeUuid()
				msg.Sender = pb.Id
				msg.Id = uuid
				msg.ChannelId = pb.Channel
				err := stream.Send(msg)
				if err != nil {
					break Lab
				}
			}
		}
		_ = conn.Close()
		time.Sleep(3 * time.Second)
	}
}