Skip to content

Commit 99a5437

Browse files
ZerpetZerpet
committed
Document manual credit flow
An example shows a somewhat complicated method to send credits every 10 chunks as manual strategy. Refactored the ProducerOptions because there were some comments not appearing in the GoDoc render. Comments in structs must be either inline or above the field.
1 parent bce9c0c commit 99a5437

File tree

3 files changed

+160
-7
lines changed

3 files changed

+160
-7
lines changed

‎examples/manual_credit/main.go‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
package main
2+
3+
import (
4+
"bufio"
5+
"log"
6+
"os"
7+
"strconv"
8+
"time"
9+
10+
"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
11+
"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
12+
)
13+
14+
func main() {
15+
done := make(chan struct{})
16+
reader := bufio.NewReader(os.Stdin)
17+
18+
l := log.New(os.Stdout, "[manual_credit] ", log.LstdFlags)
19+
l.Println("Starting Manual credit example")
20+
l.Println("Connecting to RabbitMQ streaming ...")
21+
22+
// Connect to the broker ( or brokers )
23+
env, err := stream.NewEnvironment(
24+
stream.NewEnvironmentOptions().
25+
SetHost("localhost").
26+
SetPort(5552).
27+
SetUser("guest").
28+
SetPassword("guest"))
29+
if err != nil {
30+
l.Printf("Error creating environment: %s", err)
31+
panic(err)
32+
}
33+
defer env.Close()
34+
35+
l.Println("Declaring stream")
36+
const streamName = "manual_credit"
37+
err = env.DeclareStream(streamName,
38+
&stream.StreamOptions{
39+
MaxLengthBytes: stream.ByteCapacity{}.GB(2),
40+
},
41+
)
42+
if err != nil {
43+
l.Printf("Error declaring stream: %s", err)
44+
panic(err)
45+
}
46+
defer env.DeleteStream(streamName)
47+
48+
// Get a new producer for a stream
49+
l.Println("Creating producer")
50+
opts := stream.NewProducerOptions().SetBatchSize(10_000)
51+
producer, err := env.NewProducer(streamName, opts)
52+
if err != nil {
53+
l.Printf("Error creating producer: %s", err)
54+
panic(err)
55+
}
56+
57+
// the send method automatically aggregates the messages
58+
// based on batch size
59+
l.Println("Sending 1,000,000 messages")
60+
for i := range 1_000_000 {
61+
err := producer.Send(amqp.NewMessage([]byte("hello_world_" + strconv.Itoa(i))))
62+
if err != nil {
63+
l.Printf("Error sending message: %s", err)
64+
panic(err)
65+
}
66+
}
67+
68+
time.Sleep(1 * time.Second)
69+
l.Println("Messages sent, closing producer")
70+
err = producer.Close()
71+
if err != nil {
72+
l.Printf("Error closing producer: %s", err)
73+
panic(err)
74+
}
75+
76+
count := 0
77+
receivedBatches := 0
78+
lastBatchSize := -1
79+
handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
80+
select {
81+
case <-done:
82+
return
83+
default:
84+
}
85+
if lastBatchSize == -1 {
86+
lastBatchSize = int(consumerContext.GetEntriesCount())
87+
}
88+
count++
89+
if count%10_000 == 0 {
90+
l.Printf("Received %d messages\n", count)
91+
}
92+
if count%lastBatchSize == 0 {
93+
receivedBatches++
94+
lastBatchSize = -1
95+
}
96+
if receivedBatches == 10 {
97+
l.Printf("Received %d chunks, sending 10 credits\n", receivedBatches)
98+
err := consumerContext.Consumer.Credit(10)
99+
if err != nil {
100+
l.Printf("Error sending credits: %s\n", err)
101+
}
102+
receivedBatches = 0
103+
}
104+
}
105+
106+
l.Println("Creating consumer")
107+
consumer, err := env.NewConsumer(
108+
streamName,
109+
handleMessages,
110+
stream.NewConsumerOptions().
111+
SetClientProvidedName("my_consumer"). // connection name
112+
SetConsumerName("my_consumer"). // set a consumer name
113+
SetOffset(stream.OffsetSpecification{}.First()).
114+
SetCRCCheck(false).
115+
SetCreditStrategy(stream.ManualCreditStrategy).
116+
SetInitialCredits(10)) // Initial credit
117+
if err != nil {
118+
panic(err)
119+
}
120+
121+
l.Println("Press any key to stop ")
122+
_, _, _ = reader.ReadLine()
123+
close(done)
124+
time.Sleep(200 * time.Millisecond)
125+
l.Printf("Exited after consuming %d messages\n", count)
126+
err = consumer.Close()
127+
if err != nil {
128+
l.Printf("Error closing consumer: %s\n", err)
129+
}
130+
}

‎pkg/stream/consumer.go‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,11 @@ func (cc ConsumerContext) GetEntriesCount() uint16 {
151151
return cc.chunkInfo.numEntries
152152
}
153153

154+
// MessagesHandler is the function that will be called when a message is received. This function
155+
// is used to process incoming messages. The consumer context allows access to the consumer handling
156+
// the message and some chunk information.
157+
//
158+
// When manual credit strategy is used, the consumer context allows to send credits to the broker.
154159
type MessagesHandler func(consumerContext ConsumerContext, message *amqp.Message)
155160

156161
type AutoCommitStrategy struct {
@@ -317,6 +322,16 @@ func (c *ConsumerOptions) SetSingleActiveConsumer(singleActiveConsumer *SingleAc
317322
return c
318323
}
319324

325+
// SetCreditStrategy sets the credit strategy for the consumer. Available strategies are:
326+
// - AutomaticCreditStrategy: 1 credit per chunk.
327+
// - ManualCreditStrategy: the number of credits specified in the initialCredits field.
328+
//
329+
// The credit is a flow control mechanism that allows consumers to control how RabbitMQ sends messages to them.
330+
//
331+
// The AutomaticCreditStrategy delegates the credit management to the library. The library will send 1 credit after receiving a chunk.
332+
//
333+
// The ManualCreditStrategy lets the user send credits to the broker. When ManualCreditStrategy is used, it is critically important
334+
// to send credits to keep receiving messages. The consumer context in the message handler can be used to send credits.
320335
func (c *ConsumerOptions) SetCreditStrategy(creditStrategy CreditStrategy) *ConsumerOptions {
321336
c.CreditStrategy = creditStrategy
322337
return c

‎pkg/stream/producer.go‎

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -105,16 +105,24 @@ type ProducerOptions struct {
105105
// Deduplication is a feature that allows the producer to avoid sending duplicate messages to the stream.
106106
// see: https://www.rabbitmq.com/blog/2021/07/28/rabbitmq-streams-message-deduplication for more information.
107107
// Don't use it if you don't need the deduplication.
108-
Name string
109-
QueueSize int // Internal queue to handle back-pressure.
110-
// Default value is enough high (See defaultQueuePublisherSize). You usually don't need to change it unless high memory usage is a concern.
108+
Name string
109+
// Internal queue to handle back-pressure.
110+
//
111+
// Default value is enough high (See defaultQueuePublisherSize). You usually don't need to change it unless
112+
// high memory usage is a concern.
113+
//
111114
// High value can increase the memory usage and deal with spikes in the traffic.
115+
//
112116
// Low value can reduce the memory usage but can increase the back-pressure on the server.
113-
114-
BatchSize int // It is the batch-unCompressedSize aggregation, low value reduce the latency, high value increase the throughput. Valid only for the method Send()
117+
QueueSize int
118+
// It is the batch-unCompressedSize aggregation, low value reduce the latency, high value increase the throughput.
119+
// Valid only for the method Send()
120+
BatchSize int
115121
// Deprecated: starting from 1.5.0 the SetBatchPublishingDelay is deprecated, and it will be removed in the next releases
116-
// It is not used anymore given the dynamic batching
117-
BatchPublishingDelay int // Timout within the aggregation sent a batch of messages. Valid only for the method Send()
122+
// It is not used anymore given the dynamic batching.
123+
//
124+
// Timout within the aggregation sent a batch of messages. Valid only for the method Send()
125+
BatchPublishingDelay int
118126
// Size of sub Entry, to aggregate more subEntry using one publishing id
119127
SubEntrySize int
120128
// Compression type, it is valid only if SubEntrySize > 1

0 commit comments

Comments
 (0)