main.go

package main

import (
	"context"
	"fmt"
	"time"
)

type event string
type message string

func newSource(ctx context.Context) <-chan event {
	ch := make(chan event)
	go func() {
		defer close(ch)
		ticker := time.NewTicker(time.Second)
		for {
			select {
			case <-ctx.Done():
				return
			case t := <-ticker.C:
				ch <- event(fmt.Sprintf("%s", t))
			}
		}
	}()
	return ch
}

func process(eventChan <-chan event) <-chan message {
	ch := make(chan message)
	go func() {
		defer close(ch)
		for e := range eventChan {
			ch <- message(e + ", from processor")
		}
	}()
	return ch
}

func notify(messageChan <-chan message) <-chan struct{} {
	done := make(chan struct{})
	go func() {
		defer close(done)
		for m := range messageChan {
			fmt.Println(m + ", from notifier")
		}
	}()
	return done
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	source := newSource(ctx)
	processed := process(source)
	done := notify(processed)
	<-done
}