How to call something after a channel is closed, while simultaneously being able to cancel?

[wheat tapir]

I don't know how else to phrase this, but I think it's easier if you look at the code:

func (self *fromChannel[T]) Subscribe(cx context.Context, observer Observer[T]) {

    // Receive from self.src and observer.OnNext it
    // If canceled, observer.OnComplete and return
    // If self.src is closed, observer.OnComplete and return

    // Current implementation (I don't know if it's correct)
    loop:
    for {
        select {
        case value := <-self.src:
            observer.OnNext(value)
        case <-cx.Done():
            observer.OnComplete()
            break loop
        }
        observer.OnComplete()
    }
}

Observer:

type Observer[T any] interface {
    OnNext(value T)
    OnError(err error)
    OnComplete()
}

[rich rain]

I believe that when the channel is closed, it produces the zero value to all blocked consumers. I am not 100% certain, so pls check. If I'm right, all you have to do is check if value is the zero value of T , call OnComplete and break the loop.

[indigo heart]

A channel returns 2 values where the second one is optional. The second value is a boolean and it's false when the channel is closed.

value, ok := <-self.src

If ok is false then the channel is closed.
https://go.dev/tour/concurrency/4

[wheat tapir]

thank you!