Deadletter as an actor

[perbu]

I thought I'd get a bit into the weeds and looked at what it would take to make the deadletter handling an actor (receiver).

I have made some progress, the initial stuff is there, but if possible, I'd like some feedback before I spend more time on it.

The nice thing here is that it, if we do this right, it would be simple to supply an actor to do the deadletter handling. The downside is that it seems to break quite a few assumptions in the tests. It wasn't as simple as I thought. :-)

Questions:

1. If the user is suppose to supply their own dead letter actor, how can they spawn an actor before the engine is initialized. I'm thinking maybe we could pass a producer, and the engine can initialize the actor once the engine itself is up and running.

Answer: I have an implementation in place. It's a bit hacky, see engine.go where the options are handled.

2. There is a type, DeadLetterEvent, which doesn't seem to be used much. Is the idea to encapsulate the message in this before its sent to the dead letter queue? I think that would be an improvement.

Answer: I implemented this. Messages are wrapped in DeadLetterEvent before being sent to the receiver.

3. What would be an elegant way to give the engine logger to the deadletter agent. Currently, I'm hooking it up on the startup event. What do you think about this way of doing it?

[perbu]

Ok. Some more progress.

It's now possible to pass a custom actor to deal with dead letters. It's like this:

	e := NewEngine(EngineOptDeadletter(newCustomDeadLetter))

newCustomDeadLetter returns the receiver. The engine will call this during init and set the handler. This is quite flexible, I think.

A custom deadletter receiver could look like this:

func (c *customDeadLetter) Receive(ctx *Context) {
	switch ctx.Message().(type) {
	case *DeadLetterFlush:
		c.deadLetters = c.deadLetters[:0]
	case *DeadLetterFetch:
		ctx.Respond(c.deadLetters)
	case *DeadLetterEvent:
		slog.Warn("received deadletter event")
		msg, ok := ctx.Message().(*DeadLetterEvent)
		if !ok {
			slog.Error("failed to cast deadletter event")
			return
		}
		c.deadLetters = append(c.deadLetters, msg)
	}
}

[perbu]

Quite a few tests are currently failing as a lot of the existing test code seems to touch stuff that has been changed.

[anthdm]

@perbu

1. The deadletter event is an extra (event) that is being sent each time there is message that could not be handled. So there each time there is a deadletter

a) the deadletter process will print out the msg
b) an event is being sent to the eventstream notifying actors that are interested in deadletter events. This can be handy when working in cluster.

IMO there is no way to flush the deadletter process / actor. I think it should be throttled though.

I'm looking into the failing tests. We expect (now) that when a process is not found the deadletter process will be returned, right now you are returning nil.

[perbu]

Thanks for looking into it. The reason I changed the return value to return a nil was so that we can explicitly handle the deadletter situation, so we wrap the message in a DeadLetterEvent.

I'll simplify the dead letter handling so it'll only print a warning and not keep any state. I'll also remove the Flush and Request events so we keep it dead simple.

[perbu]

wrt to b), what do you think the default behaviour should be? As it stands now deadletter is just another actor, so AFAIK there isn't any eventstream connected (unless there is some magic I don't know about).

I've not used eventstreams so far, so I'm sure what to think about it.

Should I create an manage an eventstream from the deadletter actor or should that be up the user to handle?

[anthdm]

@perbu You have a point on the event stream that is not connected though :). This means that the DeadLetter event is not going to work. Good catch. It's not super urgent to have that working though.

Good job again. I will look into the changes today and merge.

[anthdm]

@perbu I guess the merging the graceful poison, conflicts. Can you fix? :)

[perbu]

ready to merge again.