Add ees_success_ids and ees_failed_ids fields in reader config. The
former will be used to set EeIDs when the event processing returns
no error, while the latter will be used otherwise.
Add config sanity checks for the added options.
Remove Processed opts and everything related to them since they should
not be used anymore.
Fixed test compilation errors caused by the change.
Implemented functionality that handles reconnecting to the amqp server
and reinitializing the amqp channel in case of errors and timeouts. This
is handled by a goroutine created in the client constructor (it also
handles the initial connect/init).
Reconnects and reinits will use a fibonacci backoff strategy, and the
attempt amount and max waiting interval can be adjusted by the
'reconnects' and 'max_reconnect_interval' config options.
Messages that fail processing are now dropped instead of being requeued,
preventing infinite processing loops. However, this means that the
messages are lost. Handling failed messages will need to be addressed
separately.
'concurrent_requests' will now set the prefetch count. Setting the
prefetch count using the Qos function was able to replace our old
approach that was using channels. Default value is 1024 which,
according to the rabbitmq docs, 'runs into the law of diminishing
returns'. The recommended value is between 100-300. Source:
https://www.rabbitmq.com/confirms.html#channel-qos-prefetch-throughput
Fix test compilation errors and failing tests caused by these changes.
References #4160
HierarchyPath parser now returns nil when
the path is empty (instead of a string slice with one
EmptyString element).
If the prefix is set to true, when calling the AsString
method on a nil HierarchyPath, only the separator will
be returned. This avoids a nil expr error coming from
the xmlquery library.
Use the Query and QueryAll functions from the xmlquery
package to be able to handle the errors ourselves and
avoid panics.
Remove config default value for xml_root_path. The field
will remain commented in config_defaults for reference.
Add tests for HierarchyPath.AsString function.
Add comments for XmlProvider and xml_root_path opt.
Updated ees/ers implementation to use the jetstream package which
separates the jetstream context from Core NATS.
Removed the jsOpts fields from the NatsEE struct. We are now using the
jetStreamMaxWait option directly through a timeout context.
Added streamName option for NATS reader since it is now required to be
specified when creating a consumer (it is not inferred based on subject
anymore).
Updated nats ers integration test.
Removed deprecated birpc unit test.
Updated tests to also use the new jetstream package.
Updated tests to start the nats-server using their official driver
instead of using the std go exec package. time.Sleeps are now not
required anymore to wait for the server.
In test configurations for nats readers, made sure that natsStreamName
option is populated. It is now required for consumers to know where to
subscribe.
Fixed potential panic that happened when jetstreamMaxWait option would
not be set.
Upgraded NATS server version in ansible role.
upgraded go.mod nats version due to an issue caused by version mismatch
between driver and server (uncertain).
Increased amount of sleep in tests after starting a NATS server with
jetstream enabled from 50ms to 100ms.
## *NatsER.Serve
- Replaced ChanQueueSubscribe with QueueSubscribe for Core NATS consumer
to handle the message processing directly.
- Since QueueSubscribe is now used regardless of jetstream status, the
message handler has been assigned to a separate variable that can be
reused.
- The message handler is now dealing with the message processing
directly, therefore the select case listening for the channel which is
feeding NATS messages can be removed together with the channel itself
and the select. Currently, the goroutine within Serve only has to block
until the rdrExit chan is closed.
- Moved the resource check inside the handler right before starting the
message processing goroutine.
## ers.getProcessedOptions
- Renamed function from getProcessOptions to getProcessedOptions.
- Initially, the EventExporterOpts struct was always initialized to be
non-nil which meant exporting processed messages by the reader was
always enabled by force. Function has been updated to initialize it
only once when the first Processed option that was set is found.
- Created init function for all types of opts structs to avoid
repetition.
## *NatsEE.parseOpts
- Renamed function from parseOpt to parseOpts.
- Updated function to return early in case of nil opts struct to reduce
nesting.
- The nested jetstream status and maxwait conditions nil verification
have been merged into one condition.
- Handled the error coming from GetNatsOpts function.
- NATS Subject assignment has been removed. It was redundant, since it
had already been set from before this function was called.
## *NatsEE.Connect
- Updated function to return early in case of non-nil nats.Conn value
to reduce nesting.
## *NatsEE.ExportEvent
- Use defer to release resources and RUnlock.
## *NatsEE.Close
- Use defer to Unlock.
- Update function to return early in case of nil nats.Conn value to
reduce nesting.
## ees.GetNatsOpts
- Chose switch over if else when parsing client certificate and keys
opts.
- Updated function to return the errors directly instead of assigning
them to a separate variable right before returning.
## ers.GetNatsOpts
- Passed the NATSROpts struct directly to the function.
- Chose switch over if else when parsing client certificate and keys
opts.
- Updated function to return the errors directly instead of assigning
them to a separate variable right before returning.
Removed tab from commented natsJetStreamMaxWaitProcessed option
value in config_defaults.go under ers section.
Removed redundant NATS prefix from config.NATSROpts type field names.
Added function in ers/lib_test.go to create a config file in /tmp
based on a config string.
Added integration test for ERs NATS.
