example_test.go

//go:build example
// +build example

package sse

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net/http"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"

	"github.com/make-software/casper-go-sdk/sse"
)

func Test_Simple_Example(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	defer client.Stop()
	client.RegisterHandler(sse.DeployProcessedEventType, func(ctx context.Context, rawEvent sse.RawEvent) error {
		log.Printf("eventID: %d, raw data: %s", rawEvent.EventID, rawEvent.Data)
		return nil
	})
	lastEventID := 1234
	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
	defer cancel()
	assert.Error(t, client.Start(ctx, lastEventID))
}

func Test_Example_ParseDeploy(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	defer client.Stop()
	client.RegisterHandler(sse.DeployProcessedEventType, func(ctx context.Context, rawEvent sse.RawEvent) error {
		deploy, err := rawEvent.ParseAsDeployProcessedEvent()
		if err != nil {
			return err
		}
		log.Printf("Deploy hash: %s", deploy.DeployProcessed.DeployHash)
		return nil

	})
	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
	defer cancel()
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_Example_WithErrorHandling(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	defer client.Stop()
	client.ConsumerErrorHandler = func(errors <-chan error) {
		for one := range errors {
			fmt.Println(one.Error())
		}
	}
	client.RegisterHandler(sse.APIVersionEventType, TestHandler{}.ProcessEvent)
	client.RegisterHandler(sse.DeployProcessedEventType, TestHandler{}.ProcessEvent)
	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
	defer cancel()
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_Example_WithCancelContext(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	client.RegisterHandler(sse.DeployProcessedEventType, func(ctx context.Context, event sse.RawEvent) error {
		return errors.New("critical")
	})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	client.ConsumerErrorHandler = func(errors <-chan error) {
		for one := range errors {
			fmt.Println(one)
			if one.Error() == "critical" {
				cancel()
			}
		}
	}
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_Example_CloseSlowConsuming(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	client.Streamer.BlockedStreamLimit = 900 * time.Millisecond
	client.RegisterHandler(sse.BlockAddedEventType, func(ctx context.Context, event sse.RawEvent) error {
		time.Sleep(time.Second)
		return nil
	})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_Example_WithMiddleware(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	client.RegisterMiddleware(WithLogs)
	client.RegisterHandler(sse.BlockAddedEventType, TestHandler{}.ProcessEvent)
	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
	defer cancel()
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_Example_WithContextTimeout(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.TODO(), time.Second)
	defer cancel()
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	client.RegisterHandler(sse.BlockAddedEventType, TestHandler{}.ProcessEvent)
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_Example_With5Workers(t *testing.T) {
	client := sse.NewClient("http://52.3.38.81:9999/events/main")
	client.WorkersCount = 5
	client.RegisterHandler(sse.BlockAddedEventType, func(ctx context.Context, event sse.RawEvent) error {
		log.Printf("Worker: %d, eventID: %d", ctx.Value("workerID"), event.EventID)
		return nil
	})
	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
	defer cancel()
	assert.Error(t, client.Start(ctx, 1234))
}

func Test_SSE_WithConfigurations(t *testing.T) {
	// This is example how to create and configure services in idiomatic way
	streamer := sse.NewStreamer(
		sse.NewHttpConnection(
			&http.Client{
				Transport: &http.Transport{
					IdleConnTimeout:       20 * time.Minute,
					ResponseHeaderTimeout: time.Second * 30,
				},
				Timeout: 30 * time.Second,
			},
			"http://52.3.38.81:9999/events/main",
		),
		&sse.EventStreamReader{MaxBufferSize: 1024 * 1024 * 15}, // 15 MB
		2*time.Minute,
	)
	consumer := sse.NewConsumer()
	_ = streamer
	_ = consumer
}