combine-latest.go

package main

import (
	"fmt"
	"math/rand"
	"time"
)

func main() {

	c1 := asChan(1, 3, 5, 7)
	c2 := asChan(2, 4, 6, 8)

	c := combineLatest(c1, c2)
	for v := range c {
		fmt.Println(v)
	}

	time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
	for v := range c {
		fmt.Println(v)
	}
}

type Tuple struct {
	v0 interface{}
	v1 interface{}
}

func combineLatest(a <-chan interface{}, b <-chan interface{}) <-chan Tuple {

	var lasta interface{} = nil
	var lastb interface{} = nil

	c := make(chan Tuple)

	go func() {

		defer close(c)

		for a != nil || b != nil {

			select {

			case v, ok := <-a:
				if !ok {
					a = nil
					continue
				}

				lasta = v

				if lastb != nil {
					c <- Tuple{v0: v, v1: lastb}
				}

			case v, ok := <-b:
				if !ok {
					b = nil
					continue
				}

				lastb = v

				if lasta != nil {
					c <- Tuple{v0: lasta, v1: v}
				}

			}
		}
	}()

	return c
}

func asChan(vs ...int) <-chan interface{} {

	c := make(chan interface{})

	go func() {

		for i, v := range vs {

			c <- v

			if i < len(vs)-1 {
				time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
			}

		}

		close(c)
	}()

	return c
}