Creating pipelines using channels in Go

A Simple Pipeline

We start out with a book. We receive our books from a channel of type chan *Book.

type Book struct {
	Author  string
	Content string
}

func (b *Book) Print() string {
	return b.Content
}

There are two functions that are implemented against the book. One is printing the book and the second is saving the book into some sort of a data store. We will simplify that by just println. Note that save is our final step at which our program terminates if there are no more books (i.e. channel is closed)

func print(input chan *Book) chan *Book {
	output := make(chan *Book)
	go func() {
		defer close(output)
		for printable := range input {
			println("Printing:", printable.Print())
			output <- printable
		}
	}()
	return output
}

func saveBook(input chan *Book) {
	for book := range input {
		println("Saving book for author:", book.Author)
	}
}

We will simulate the input channel as

books := make(chan *Book)

go func() {
	defer close(books)
	for i := 0; i <= 1; i++ {
		books <- &Book{fmt.Sprintf("Author-%d", i), fmt.Sprintf("Content-%d", i)}
	}
}()

Our implementation is straight forward, and actually very elegant.

saveBook(print(books))

Yes, that’s it. Every step will do its thing then hand the book to the next step.

Working example in Go Playground

Printing Books and Magazines

Our printing house is having a new business opportunity. We are printing a Magazine

type Magazine struct {
	Author  string
	Content string
	Issue int
}

func (m *Magazine) Print() string {
	return m.Content
}

Magazine has its own specific saveMagazine function

func saveMagazine(magazines chan *Magazine) {
	for magazine := range magazines {
		println("Saving magazine for issue:", magazine.Issue)
	}
}

Here comes the tricky part. Whether it is a magazine or a book, we are using the same print mechanism. But our print(input chan *Book) chan *Book is useful only for books. We can copy and paste all our printing business, but that doesn’t sound like the best idea.

print doesn’t care if it is a magazine or a book, as long as it has a Print function. This brings to mind the keyword interface. Let’s create a Printable interface and make print depend on that instead.

type Printable interface {
	Print() string
}

func print(input chan Printable) chan Printable {
	output := make(chan Printable)
	go func() {
		defer close(output)

		for printable := range input {
			println("Printing:", printable.Print())
			output <- printable
		}
	}()
	return output

}

This looks better. A print function only cares about printing. It doesn’t care about authors or issues! But Go will not get away so easy with it. Go wouldn’t let us cast channels. var books chan Printable = make(chan *Book) is illegal. There is no casting to make this happen and we will need to write our own conversion for any client that wants to use this function.

Let’s focus on books and magazine will be the same. We will create two conversion functions back and forth between Printable and Book channels

func bookChanToPrintableChan(books chan *Book) (chan Printable) {
	printables := make(chan Printable)
	go func() {
		defer close(printables)
		for printable := range books {
			printables <- printable
		}
	}()
	return printables
}

func printableChanToBookChan(printables chan Printable) (chan *Book) {
	books := make(chan *Book)
	go func() {
		defer close(books)
		for printable := range printables {
			books <- printable.(*Book)
		}
	}()
	return books
}

our pipeline will look something like this:

saveBook(printableChanToBookChan(print(bookChanToPrintableChan(books))))

Working example in Go Playground

Thinking Outside The Box

We might not be able to cast from a channel of one type to a channel of another. But we can cast channels to a compatible type. Let’s see what we can do with that.

type Books chan *Book

func (b Books) Printables() chan Printable{
	printables := make(chan Printable)
	go func() {
		defer close(printables)
		for printable := range b {
			printables <- printable
		}
	}()
	return printables
}

type Printables chan Printable

func (p Printables) Books() chan *Book {
	books := make(chan *Book)
	go func() {
		defer close(books)
		for printable := range p {
			books <- printable.(*Book)
		}
	}()
	return books
}

We created two compatible types Books and Printables. Each of them has a method that converts them into the other type. If we had a magazine Printable would have had another method Magazines. Now we can cast chan *Book to Books. Our code can be rewritten using casting.

saveBook(Printables(print(Books(books).Printables())).Books())

Working example in Go Playground

There is still final compromise with declaring new types and using casting. That is using the new declared type directly. The downside with that, is we bring back a bit of coupling. On the channel type declaration, we are letting the Printables channel know about Books channel, which shouldn’t be its concern. However, I just brought that in to make all options available.

type Books chan *Book

func (b Books) Printables() Printables {
	printables := make(Printables)
	go func() {
		defer close(printables)
		for printable := range b {
			printables <- printable
		}
	}()
	return printables
}

type Printables chan Printable

func (p Printables) Books() Books {
	books := make(Books)
	go func() {
		defer close(books)
		for printable := range p {
			books <- printable.(*Book)
		}
	}()
	return books
}

Now, save book will look something like this:

saveBook(print(books.Printables()).Books())

Working example in Go Playground

These are the three methods side by side. I think with better naming the first example looks more understandable since it highlights the left to right relationship. The last example is not too bad, except for the fact that we still have coupling.

saveBook(printableChanToBookChan(print(bookChanToPrintableChan(books))))
saveBook(Printables(print(Books(books).Printables())).Books())
saveBook(print(books.Printables()).Books())

The choice of which approach to take, will mostly depend on how many of these conversions one might have?