Channels and Pipelines

Channel[T] connects concurrent producers and consumers with explicit sealing behavior.

Channel[T]

channel(capacity) creates a bounded channel for values of type T.

func main(args: List[String]) -> Void:
	ch: Channel[Int] = channel(10)

Send/Recv

send blocks when the channel is full; recv returns Option[T] and returns None when the channel is sealed and drained.

_ = send(ch, 42)
value: Option[Int] = recv(ch)

Seal

seal tells receivers that no more values will arrive. close remains as a compatibility alias.

seal(ch)
match recv(ch):
    None: print("sealed")
    Some(v): print(v)

For-In Over Channels

A for loop over a channel receives values until the channel is sealed and drained.

for value in ch:
    total += value

Example

pipeline.brp

func produce(ch: Channel[Int], start: Int, end: Int) -> Void:
	var i: Int = start
	while i < end:
		_ = send(ch, i)
		i += 1


func consume(ch: Channel[Int]) -> Int:
	var total: Int = 0
	for n in ch:
		total += n
	total


func main(args: List[String]) -> Int:
	ch: Channel[Int] = channel(10)
	concurrent:
		first = produce(ch, 0, 3)
		second = produce(ch, 3, 5)
	seal(ch)
	print(consume(ch)) -- prints: 10
	0

Try it

terminal

blorp run pipeline.brp