Captures a panic from calling Map

rxgo_panic_handler.go

package main

import (
	"context"
	"fmt"
	"github.com/reactivex/rxgo/v2"
)

type MapFunc = func(ctx context.Context, i interface{}) (interface{}, error)

func MapWrapper(mapFunc MapFunc) MapFunc {
	return func(ctx context.Context, i interface{}) (interface{}, error) {
		resultChan := make(chan interface{})
		errChan := make(chan error)

		go func() {
			defer func() {
				if r := recover(); r != nil {
					errChan <- fmt.Errorf("panic: %v", r)
				}
			}()
			if result, err := mapFunc(ctx, i); err != nil {
				errChan <- err
			} else {
				resultChan <- result
			}
		}()

		select {
		case res := <-resultChan:
			return res, nil
		case err := <-errChan:
			return nil, err
		}
	}
}

func main() {
	observable := rxgo.Just("foo")().
		Map(func(ctx context.Context, i interface{}) (interface{}, error) {
			fmt.Println("map 1: {yawn!}")
			return i, nil
		}).
		Map(MapWrapper(func(ctx context.Context, i interface{}) (interface{}, error) {
			fmt.Println("map 2: pretend this is calling an API that panics")
			panic("well fuck.")
		}))

	<-observable.ForEach(
		func(i interface{}) {
			fmt.Printf("ForEach Item: %v\n", i)
		},
		func(err error) {
			fmt.Printf("ForEach Error: %v\n", err)
		},
		func() {
			fmt.Println("ForEach Completed.")
		},
	)

	fmt.Println("Entering loop...")
	<-make(chan struct{})
}

/**
map 1: {yawn!}
map 2: pretend this is calling an API that panics
ForEach Error: panic: well fuck.
ForEach Completed.
Entering loop...
*/