package main import ( "context" "fmt" "os" "os/signal" "strconv" "time" "github.com/google/uuid" "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp" "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream" ) func CheckErr(err error) { if err != nil { fmt.Printf("%s ", err) os.Exit(1)	} } func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill) defer stop() env, err := stream.NewEnvironment( stream.NewEnvironmentOptions(). SetHost("localhost"). SetPort(5552). SetUser("guest"). SetPassword("guest").SetRequestedHeartbeat(time.Second * 1)) CheckErr(err) streamName := uuid.New().String() err = env.DeclareStream(streamName, &stream.StreamOptions{ MaxLengthBytes: stream.ByteCapacity{}.MB(500),	},	) CheckErr(err) consName := "consumer_" producer, err := env.NewProducer(streamName, nil) CheckErr(err) go func() { for i := 0; i < 1000; i++ { err := producer.Send(amqp.NewMessage([]byte("hello_world_" + strconv.Itoa(i)))) CheckErr(err) time.Sleep(5 * time.Second)	}	}() handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) { err := consumerContext.Consumer.StoreCustomOffset(consumerContext.Consumer.GetOffset()) if err != nil { CheckErr(err)	} fmt.Println(string(message.GetData()))	} cons, err := env.NewConsumer(streamName, handleMessages, stream. NewConsumerOptions(). SetManualCommit(). SetOffset(stream.OffsetSpecification{}.LastConsumed()).SetConsumerName(consName),	) CheckErr(err) defer cons.Close() <-ctx.Done() }