---
title: "Drain Internal Channels"
description: "Finish a producer/consumer Flow after its internally published Channel messages have been processed."
canonical_url: "https://docs.superdurable.io/design-patterns/drain-internal-channels/"
source: "docs/content/design-patterns/drain-internal-channels.mdx"
locale: "en"
---

# Drain Internal Channels

Finish a producer/consumer Flow after its internally published Channel messages have been processed.

## Problem and approach

Use this pattern when one branch of a Flow publishes work for another branch. **Init** starts **Main** and **Side** in parallel. **Main** publishes each document command to **SideStepData**. **Side** waits for one Channel value, writes it, and waits again.

The key is the sentinel published by **Finalize**. After **Main** has published every ordinary message, **Finalize** publishes a final command to **SideStepData**. Because a Channel is a FIFO queue, **Side** cannot receive that sentinel until it has received every earlier message. It completes only when it receives the sentinel, so the Flow does not finish before every published document is handled. With several producer branches, make **Finalize** wait for all producers before it publishes the sentinel.

**Flow: DrainInternalChannelFlow**

Start Step: `Init`

**Steps**

- `Finalize`
  - `gracefulComplete`
- `Init`
- `Main`
- `Side`
  - WaitFor `until`: channel `SideStepData.for 1`
  - `gracefulComplete` when `document.final_command`

**Edges**

- `goToMany` (decision; Init) → `Main` (step): fan-out
- `goToMany` (decision; Init) → `Side` (step): fan-out
- `goTo` (decision; Main; when `execution_count <= 3`) → `Main` (step): go_to
- `goTo` (decision; Main; when `not (execution_count <= 3)`) → `Finalize` (step): go_to
- `goTo` (decision; Side; when `not (document.final_command)`) → `Side` (step): go_to
- `main_step_execution_counter` (attribute) → `Main` (step): get
- `SideStepData` (channel) → `Side` (step): results
- `SideStepData` (channel) → `until` (wait; Side): SideStepData.for 1
- `Finalize` (step) → `SideStepData` (channel): publish
- `Init` (step) → `main_step_execution_counter` (attribute): set
- `Main` (step) → `main_step_execution_counter` (attribute): set
- `Main` (step) → `SideStepData` (channel): publish

Example: [examples/python/dex_examples/patterns/drain-channels/internal/drain_internal_channels_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/patterns/drain-channels/internal/drain_internal_channels_flow.py)

## Code examples

**Python**

```python
class Main(Step[str]):
    def execute(self, context: Context, input: str) -> StepDecision:
        execution_count = self.execution_counter.get(context) + 1
        self.execution_counter.set(context, execution_count)
        statuses = {1: "RECEIVED", 2: "ACCEPTED", 3: "PASSED"}
        self.side_step_data.publish(
            context,
            MongoDocument(input, statuses.get(execution_count, "ERROR"), False),
        )
        if execution_count <= 3:
            return go_to(Main, input)
        return go_to(Finalize, None)

class Side(Step[None]):
    def wait_for(self, context: Context, input: None) -> Wait:
        return Wait.until(self.side_step_data.for_one())

    def execute(self, context: Context, input: None) -> StepDecision:
        document = self.side_step_data.results(context)[0]
        self.mongo_collection.upsert(document)
        if document.final_command:
            return graceful_complete()
        return go_to(Side, None)

class Finalize(Step[None]):
    def execute(self, context: Context, input: None) -> StepDecision:
        self.side_step_data.publish(
            context, MongoDocument("documentId-1", "FINALIZED", True)
        )
        return graceful_complete()
```

Example: [examples/python/dex_examples/patterns/drain-channels/internal/drain_internal_channels_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/patterns/drain-channels/internal/drain_internal_channels_flow.py)

**Go**

```go
func (step mainStep) Execute(ctx dex.Context, input string) (*dex.StepDecision, error) {
	executionCount, err := MainStepExecutionCounterAttribute.Get(ctx)
	if err != nil {
		return nil, err
	}
	executionCount++
	if err := MainStepExecutionCounterAttribute.Set(ctx, executionCount); err != nil {
		return nil, err
	}
	status := "ERROR"
	switch executionCount {
	case 1:
		status = "RECEIVED"
	case 2:
		status = "ACCEPTED"
	case 3:
		status = "PASSED"
	}
	if err := SideStepData.Publish(ctx, MongoDocument{
		ID:           input,
		Status:       status,
		FinalCommand: false,
	}); err != nil {
		return nil, err
	}
	if executionCount <= 3 {
		return dex.GoTo(mainStep{}, input), nil
	}
	return dex.GoTo(finalizeStep{}, nil), nil
}

func (sideStep) WaitFor(dex.Context, dex.None) (*dex.Wait, error) {
	return dex.Until(SideStepData.ForOne()), nil
}

func (step sideStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecision, error) {
	documents, err := SideStepData.GetConditionResults(ctx)
	if err != nil {
		return nil, err
	}
	document := documents[0]
	if err := step.service.Upsert(document); err != nil {
		return nil, err
	}
	if document.FinalCommand {
		return dex.GracefulComplete(nil), nil
	}
	return dex.GoTo(sideStep{}, nil), nil
}

func (finalizeStep) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecision, error) {
	if err := SideStepData.Publish(ctx, MongoDocument{
		ID: "documentId-1", Status: "FINALIZED", FinalCommand: true,
	}); err != nil {
		return nil, err
	}
	return dex.GracefulComplete(nil), nil
}
```

Example: [examples/go/patterns/drain-channels/internal-drain/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/patterns/drain-channels/internal-drain/workflow.go)

**Java**

```java
final class Main implements Step<String> {
    public StepDecision execute(final Context context, final String input) {
    final int executionCount = mainStepExecutionCounter.get(context) + 1;
    mainStepExecutionCounter.set(context, executionCount);

    final String status;
    switch (executionCount) {
        case 1:
            status = "RECEIVED";
            break;
        case 2:
            status = "ACCEPTED";
            break;
        case 3:
            status = "PASSED";
            break;
        default:
            status = "ERROR";
            break;
    }
    sideStepData.publish(
            context,
            new MongoDocument(input, status, false));
    if (executionCount <= 3) {
        return StepDecision.goTo(Main.class, input);
    }
    return StepDecision.goTo(Finalize.class, null);
    }
}

final class Side implements Step<Void> {
public Wait waitFor(final Context context, final Void input) {
    return Wait.until(sideStepData.forOne());
}

public StepDecision execute(final Context context, final Void input) {
    final MongoDocument document = sideStepData.getConditionResults(context).get(0);
    mongoCollection.upsert(document);
    if (document.finalCommand) {
        return StepDecision.gracefulComplete();
    }
    return StepDecision.goTo(Side.class, null);
}
}

final class Finalize implements Step<Void> {
public StepDecision execute(final Context context, final Void input) {
    sideStepData.publish(
            context,
            new MongoDocument("documentId-1", "FINALIZED", true));
    return StepDecision.gracefulComplete();
}
}
```

Example: [examples/java/src/main/java/io/superdurable/dex/patterns/drainchannels/internal/DrainInternalChannelFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/patterns/drainchannels/internal/DrainInternalChannelFlow.java)

**TypeScript**

```typescript
public execute(context: Context, input: string): StepDecision {
  const executionCount = this.flow.mainStepExecutionCounter.get(context) + 1;
  this.flow.mainStepExecutionCounter.set(context, executionCount);

  let status: string;
  switch (executionCount) {
    case 1:
      status = "RECEIVED";
      break;
    case 2:
      status = "ACCEPTED";
      break;
    case 3:
      status = "PASSED";
      break;
    default:
      status = "ERROR";
      break;
  }

  sideStepData.publish(context, {
    id: input,
    status,
    finalCommand: false,
  });
  if (executionCount <= 3) {
    return goTo(Main, input);
  }
  return goTo(Finalize, undefined);
}

public waitFor(_context: Context, _input: void): Wait {
  return Wait.until(sideStepData.forOne());
}

public execute(context: Context, _input: void): StepDecision {
  const document = sideStepData.results(context)[0];
  this.flow.mongoCollection.upsert(document);
  if (document.finalCommand) {
    return gracefulComplete();
  }
  return goTo(Side, undefined);
}

public execute(context: Context, _input: void): StepDecision {
  sideStepData.publish(context, {
    id: "documentId-1",
    status: "FINALIZED",
    finalCommand: true,
  });
  return gracefulComplete();
}
```

Example: [examples/typescript/src/patterns/drain-channels/internal/drain-internal-channels-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/patterns/drain-channels/internal/drain-internal-channels-flow.ts)

**Rust**

```rust
impl Step for Main {
    type Input = Vec<String>;

    fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult<StepDecision> {
        for value in input {
            side_step_data().publish(context, SideStepData::Message(value))?;
        }
        Ok(StepDecision::go_to(&Finalize, ()))
    }
}

impl Step for Side {
    type Input = ();

    fn wait_for(&self, _context: &mut Context, _input: ()) -> HandlerResult<Wait> {
        Ok(Wait::until(side_step_data().for_one()))
    }

    fn execute(&self, context: &mut Context, _input: ()) -> HandlerResult<StepDecision> {
        let command = side_step_data()
            .condition_results(context)?
            .into_iter()
            .next()
            .unwrap_or_default();
        match command {
            SideStepData::Message(value) => {
                context.record_event("drained-internal", value)?;
                Ok(StepDecision::go_to(&Side, ()))
            }
            SideStepData::Final => Ok(StepDecision::graceful_complete(())),
        }
    }
}

impl Step for Finalize {
    type Input = ();

    fn execute(&self, context: &mut Context, _input: ()) -> HandlerResult<StepDecision> {
        side_step_data().publish(context, SideStepData::Final)?;
        Ok(StepDecision::graceful_complete(()))
    }
}
```

Example: [examples/rust/src/patterns/drain_channels/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/patterns/drain_channels/flow.rs)

## Related

- [Channel](https://docs.superdurable.io/primitives/channel.md)
- [Drain Channels](https://docs.superdurable.io/design-patterns/drain-channels.md)