Niclas,
Il giorno mar 24 gen 2023 alle ore 09:04 Niclas Hedhman
<nic...@hedhman.org> ha scritto:
>
> Hi,
> I am trying to set up topics to act as "message buffers" to web clients.
>
> I would like the client to be able to re-connect at any time, and have
> the latest N messages (or AFAIUI, some number of Megabytes).
>
> So, I think I need;
> 1. non-persistent topics
> 2. very long MessageTTL on Namespace used for this
> 3. RetentionPolicies with -1 time limit and some number of MB
>
> Client in question is GO.
>
> I am reading (see below) the topic, not subscribing.
>
> Question 1;
> When I do 'pulsar.EarliestMessageID()', should I be getting the oldest
> message in the Retention buffer, or the oldest message not acknowledged?
>
> If the latter, will subsequent messages be read out, or will the same
> client be served the same message over and over again?
>
> Question 2;
> Will RetentionPolicies ("size"/"time") cause messages to be kept in RAM,
> or can I rely on messages being read from disk, even though they are
> non-persistent?
A non-persistent topic is volatile: if you restart the broker you lose the data
Enrico
>
>
> TIA
> Niclas
>
>
> -o-o-o- Relevant code -o-o-o-
> Creating reader;
>
> func (p *PulsarClient) CreateReader(topic string, earliest bool)
> pulsar.Reader {
> var start pulsar.MessageID
> if earliest {
> start = pulsar.EarliestMessageID()
> } else {
> start = pulsar.LatestMessageID()
> }
> reader, err := p.client.CreateReader(pulsar.ReaderOptions{
> Topic: topic,
> StartMessageID: start,
> })
> if err != nil {
> log.DefaultLogger.Error(fmt.Sprintf("Failed to create Pulsar
> Reader for: %s", topic), err)
> }
> return reader
> }
>
>
> Reading messages;
> reader :=
> h.pulsar.CreateReader(model.NotificationTopics+strconv.FormatInt(orgId,
> 10), true)
> defer reader.Close()
> for {
> msg, err := reader.Next(ctx)
> if msg == nil {
> log.DefaultLogger.Info("Grafana sender: DONE")
> return ctx.Err()
> }
> if err != nil {
> log.DefaultLogger.Error(fmt.Sprintf("Couldn't get the
> message via reader.Next(): %+v", err))
> continue
> }
> log.DefaultLogger.Info("Sending notification to " +
> req.PluginContext.User.Login)
> err = sender.SendJSON(msg.Payload())
> if err != nil {
> log.DefaultLogger.Error(fmt.Sprintf("Couldn't send frame:
> %v", err))
> return err
> }
> }
> }
