Completar, cancelar y liberar un hub

Completar permite terminar de leer; disponer aborta las suscripciones. Elige según necesites drenar o detener.

using Greencore.platform.streaming;

await using var hub = StreamHub<int>.CreateWorkQueue();
await using var suscripcion = hub.Subscribe();
var consumidor = LeerAsync(suscripcion);
await hub.PublishAsync(10);
hub.Complete();
await consumidor;
var cierre = await suscripcion.Completion;
Console.WriteLine(cierre.Kind); // Completed.

static async Task LeerAsync(StreamSubscription<int> suscripcion)
{
    await foreach (var numero in suscripcion.ReadAllAsync())
        Console.WriteLine(numero);
}

Alcance de cada operación

Operación	Alcance
hub.Complete()	Impide publicar y permite drenar
hub.Fail(exception)	Permite drenar y luego propaga el fallo
suscripcion.CancelAsync()	Cancela y espera la limpieza de esa suscripción
suscripcion.DisposeAsync()	Libera esa suscripción
hub.DisposeAsync()	Aborta suscripciones y espera la limpieza interna

También puedes pasar un token a Subscribe(token) o ReadAllAsync(token): cancela esa suscripción. La primera terminación del productor prevalece; después del cierre no se admiten nuevas suscripciones.

suscripcion.Completion informa Completed, Failed, Canceled o Disposed, y conserva la excepción cuando corresponde. Una suscripción nunca enumerada debe cancelarse o disponerse explícitamente.

Recuerda: disponer el hub no espera el trabajo dentro de tus bucles. Conserva las tareas de los consumidores y espéralas; para abortar su procesamiento, pasa también un token a ese trabajo.