usedatabrew / tango Goto Github PK

package main

import (
	"fmt"
	"time"

	"github.com/usedatabrew/tango"
)

func main() {
	instance := tango.NewTango()

	stages := []tango.Stage{
		{
			Channel: make(chan interface{}),
			Function: func(msg interface{}) (interface{}, error) {
				return fmt.Sprintf("%v stage 1", msg), nil
			},
		},
		{
			Channel: make(chan interface{}),
			Function: func(msg interface{}) (interface{}, error) {
				if msg.(int)%5 == 0 {
					time.Sleep(time.Millisecond * 10)
				}
				return fmt.Sprintf("%v stage 2", msg), nil
			},
		},
		{
			Channel: make(chan interface{}),
			Function: func(msg interface{}) (interface{}, error) {
				return fmt.Sprintf("%v stage 3", msg), nil
			},
		},
	}

	instance.SetStages(stages)

	instance.OnProcessed(func(i interface{}, err error) {
		if err != nil {
			_ = fmt.Errorf("messages happened in the last stage %v", err)
		}
		fmt.Println("Message processed", i)
	})

	producerChannel := make(chan interface{})

	go func() {
		for i := 0; i <= 10000000; i++ {
			producerChannel <- fmt.Sprintf("item %v", i)
			time.Sleep(time.Millisecond * 200)
		}
	}()

	// Tango will consume messages from the producer channel and
	// pass them through the stages.
	instance.SetProducerChannel(producerChannel)

	if err := instance.Start(); err != nil {
		panic(err)
	}
}