(ctx context.Context, configMap ConfigMap)
| 71 | } |
| 72 | |
| 73 | func NewPulsarEventQueueFactory(ctx context.Context, configMap ConfigMap) (TubeFactory, error) { |
| 74 | config, err := NewPulsarTubeFactoryConfig(configMap) |
| 75 | if err != nil { |
| 76 | return nil, err |
| 77 | } |
| 78 | pc, err := pulsar.NewClient(pulsar.ClientOptions{ |
| 79 | URL: config.PulsarURL, |
| 80 | }) |
| 81 | if err != nil { |
| 82 | return nil, err |
| 83 | } |
| 84 | var closed atomic.Bool // TODO: Remove this after the bug of Producer.Flush is fixed |
| 85 | go func() { |
| 86 | <-ctx.Done() |
| 87 | slog.InfoContext(ctx, "Closing Pulsar event queue factory", slog.Any("config", config)) |
| 88 | closed.Store(true) |
| 89 | pc.Close() |
| 90 | }() |
| 91 | handleErr := func(ctx context.Context, err error, message string, args ...interface{}) { |
| 92 | if errors.Is(err, context.Canceled) { |
| 93 | slog.InfoContext(ctx, "Pulsar queue cancelled", slog.Any("config", config)) |
| 94 | return |
| 95 | } |
| 96 | extraArgs := append(args, slog.Any("config", config), slog.Any("error", err)) |
| 97 | slog.ErrorContext(ctx, message, extraArgs...) |
| 98 | } |
| 99 | log := func(message string, config interface{}, args ...interface{}) { |
| 100 | slog.InfoContext(ctx, message, append(args, slog.Any("config", config))...) |
| 101 | } |
| 102 | return &PulsarEventQueueFactory{ |
| 103 | newSourceChan: func(ctx context.Context, config *SourceQueueConfig) (<-chan Record, error) { |
| 104 | c := make(chan Record) |
| 105 | consumer, err := pc.Subscribe(pulsar.ConsumerOptions{ |
| 106 | Topics: config.Topics, |
| 107 | SubscriptionName: config.SubName, |
| 108 | Type: pulsar.Failover, |
| 109 | }) |
| 110 | if err != nil { |
| 111 | return nil, errors.Wrap(err, "Error creating consumer") |
| 112 | } |
| 113 | log("Pulsar source queue created", config) |
| 114 | go func() { |
| 115 | defer log("Pulsar source queue closed", config) |
| 116 | defer consumer.Close() |
| 117 | defer close(c) |
| 118 | for msg := range consumer.Chan() { |
| 119 | c <- NewRecordImpl(msg.Payload(), func() { |
| 120 | err := consumer.Ack(msg) |
| 121 | if err != nil { |
| 122 | handleErr(ctx, err, "Error acknowledging message", "error", err) |
| 123 | return |
| 124 | } |
| 125 | }) |
| 126 | } |
| 127 | }() |
| 128 | return c, nil |
| 129 | }, |
| 130 | newSinkChan: func(ctx context.Context, config *SinkQueueConfig) (chan<- Record, error) { |
no test coverage detected