Листинги файлов

cmd/consumer/main.go

 
package mainimport (	"context"	"fmt"	"os/signal"	"syscall"	"time"	"github.com/google/uuid"	"github.com/vmihailenco/msgpack/v5"	"gitlab.vkteam.ru/tarantool/tqe/message-queue.git/v3/internal/protocol"	"google.golang.org/grpc"	"google.golang.org/grpc/credentials/insecure")type ExamplePayload struct {	Foo string	Bar string}func main() {	ctx := context.Background()	conn, err := grpc.DialContext(		ctx,		"localhost:18182",		grpc.WithTransportCredentials(insecure.NewCredentials()),	)	if err != nil {		panic(err)	}	defer conn.Close()	client := protocol.NewConsumerClient(conn)	subscribe, err := client.Subscribe(ctx)	if err != nil {		panic(err)	}	req := &protocol.SubscriptionRequest{}	req.SetQueue("queue")	req.SetConsumeId("example-consume-id-" + uuid.NewString())	req.SetTtl(float32(time.Minute.Seconds()))	sreq := new(protocol.SubscriptionStreamRequest)	sreq.SetSubscribeRequest(req)	if err := subscribe.Send(sreq); err != nil {		panic(err)	}	fmt.Println("Start listening...")	go func() {		for {			recv, err := subscribe.Recv()			if err != nil {				panic(err)			}			switch recv.WhichResponse() {			case protocol.SubscriptionStreamResponse_Notifications_case:				notifResp := recv.GetNotifications()				if notifResp == nil {					panic("returned empty list of notifications")				}				fmt.Printf("Received %d messages\n", len(notifResp.GetNotifications()))				for _, notification := range notifResp.GetNotifications() {					example := ExamplePayload{}					err = msgpack.Unmarshal(notification.GetMessage().GetPayload(), &example)					if err != nil {						panic(err)					}					fmt.Printf("foo: %s bar: %s\n", example.Foo, example.Bar)					cursor := notification.GetCursor()					comreq := new(protocol.CommitRequest)					comreq.SetCursor(cursor)					streamreq := new(protocol.SubscriptionStreamRequest)					streamreq.SetCommitRequest(comreq)					if err := subscribe.Send(streamreq); err != nil {						panic(err)					}				}			case protocol.SubscriptionStreamResponse_CommitResponse_case:				resp := recv.GetCommitResponse()				cursor := resp.GetCursor()				fmt.Printf("Consumer state has updated by %s cursor\n", cursor)			default:			}		}	}()	notifyCtx, _ := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)	<-notifyCtx.Done()	fmt.Println("exit")}

cmd/producer/main.go:

 
package mainimport (	"context"	"fmt"	"time"	"github.com/vmihailenco/msgpack/v5"	"gitlab.vkteam.ru/tarantool/tqe/message-queue.git/v3/internal/protocol"	"google.golang.org/grpc"	"google.golang.org/grpc/credentials/insecure")type ExamplePayload struct {	Foo string	Bar string}func main() {	ctx := context.Background()	conn, err := grpc.DialContext(		ctx,		"localhost:18182",		grpc.WithTransportCredentials(insecure.NewCredentials()),	)	if err != nil {		panic(err)	}	defer conn.Close()	client := protocol.NewProducerClient(conn)	payload, err := msgpack.Marshal(&ExamplePayload{		Foo: "ooF",		Bar: time.Now().Format(time.RFC3339Nano),	})	if err != nil {		panic(err)	}	pair := new(protocol.Pair)	pair.SetKey([]byte("Content-Type"))	pair.SetValue([]byte("application/protobuf"))	batchReq := &protocol.ProduceRequest{}	batchReq.SetQueue("queue")	msg1 := &protocol.ProduceMessage{}	msg1.SetRoutingKey([]byte("rk2"))	msg1.SetPayload(payload)	msg1.SetMetadata([]*protocol.Pair{pair})	msg2 := &protocol.ProduceMessage{}	msg1.SetRoutingKey([]byte("rk1"))	msg2.SetPayload(payload)	msg2.SetMetadata([]*protocol.Pair{pair})	batchReq.SetMessages([]*protocol.ProduceMessage{msg1, msg2})	responseBatch, err := client.Produce(ctx, batchReq)	if err != nil {		panic(err)	}	fmt.Printf("Messages batch published with ids: %v\n", responseBatch.GetIds())}