exgo with gocron example

rxgo_with_gocron.go

package main

import (
	"context"
	"fmt"
	"github.com/go-co-op/gocron"
	"github.com/reactivex/rxgo/v2"
	"time"
)

func main() {

	ch := make(chan rxgo.Item)

	s := gocron.NewScheduler(time.UTC)
	_, _ = s.Every(10).Seconds().Do(func() {
		data := time.Now().UnixNano() / 1000 % 100
		fmt.Printf("writing data to channel: %v\n", data)
		ch <- rxgo.Of(data)
	})

	s.StartAsync()

	observable := rxgo.FromChannel(ch).
		Map(func(ctx context.Context, i interface{}) (interface{}, error) {
			//fmt.Println(i.(int))
			value := i.(int64) * 2
			return value, nil
		}).
		Filter(func(i interface{}) bool {
			return i.(int64) < 50
		})

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()

	observable.Connect(ctx)

	for items := range observable.Observe() {
		fmt.Println("data received :: ", items.V)
	}
}