Remove bespoke SendMessage implementation

[dan-j]

Problem
Copying the kncloudevents.Dispatcher smells a bit, and I have a proposal to remove it.

We can implement our own WithRetryConfig and WithDeadLetterSink options which control the dispatcher based on current state. The one thing I have a solution for, but 50/50 if I like it, is signalling when to nak/term messages...

I have this bit of code in WIP:

func ReportWithoutRetry(result *error, attempt int, cfg *kncloudevents.RetryConfig) *kncloudevents.RetryConfig {
	if cfg == nil {
		return nil
	}

	next := *cfg

	next.CheckRetry = func(ctx context.Context, resp *http.Response, err error) (bool, error) {
		retry, err := cfg.CheckRetry(ctx, resp, err)

		if retry {
			if cfg.Backoff != nil {
				multierr.AppendInto(result, NewDelayNak(cfg.Backoff(attempt, resp)))
			} else {
				multierr.AppendInto(result, protocol.ResultNACK)
			}
		}

		return false, err
	}

	return &next
}

We take in the RetryConfig defined on the subscription, and decorate it with our own CheckRetry function which always returns false but sets a protocol.ResultNACK error. You can then use it like so:

var retryError error

defer func() {
	err = msg.Finish(multierr.Append(err, retryError))
}()

...

dispatchExecutionInfo, err := c.dispatcher.SendMessage(
	ctx,
	internal.WithoutFinish(msg),
	c.sub.Subscriber,
	internal.WithRetryReporter(&retryError, attempt, c.sub.RetryConfig),
	internal.WithDeadLetterOnLastAttempt(attempt, c.sub.RetryConfig, c.sub.DeadLetter),
	kncloudevents.WithReply(c.sub.Reply),
	kncloudevents.WithTransformers(&te),
)

The finish method of the message (see #542) then understands how to check the errors to determine whether to nack, and if so what delay.

At the moment, my dead-letter handling will just ack the message. I'm not sure if this should result in a Term or not?

Persona:

• System Integrator

Exit Criteria
No copy/paste of kncloudevents.

Time Estimate (optional):
1 day

[github-actions]

This issue is stale because it has been open for 90 days with no
activity. It will automatically close after 30 more days of
inactivity. Reopen the issue with /reopen. Mark the issue as
fresh by adding the comment /remove-lifecycle stale.

[astelmashenko]

/remove-lifecycle stale