multiplexedsource_test.go

// Copyright 2019 dfuse Platform Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package bstream

import (
	"fmt"
	"testing"
	"time"

	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestMultiplexedSource(t *testing.T) {
	sourceReconnectDelay = time.Duration(0)

	sfOne := NewTestSourceFactory()
	sfTwo := NewTestSourceFactory()

	doneCount := 0
	done := HandlerFunc(func(blk *pbbstream.Block, obj any) error {
		doneCount++
		return nil
	})

	mplex := NewMultiplexedSource([]SourceFactory{sfOne.NewSource, sfTwo.NewSource}, done)
	go mplex.Run()
	srcOne := <-sfOne.Created
	srcTwo := <-sfTwo.Created

	<-srcOne.running // test fixture ready to push blocks
	<-srcTwo.running // test fixture ready to push blocks

	now := time.Now().UTC()

	require.NoError(t, srcOne.Push(TestBlockWithTimestamp("00000002a", "00000001a", now), nil))
	require.Equal(t, 1, doneCount)

	srcTwo.Shutdown(fmt.Errorf("test"))
	<-srcTwo.Terminating()

	srcTwo = <-sfTwo.Created
	require.NoError(t, srcTwo.Push(TestBlockWithTimestamp("00000004a", "00000003a", now), nil))

	require.Equal(t, 2, doneCount, "sending new block from reconnected source")

	mplex.Shutdown(nil)
	<-srcOne.Terminating()
	<-srcTwo.Terminating()
}

func TestMultiplexedSource_shutdownOnProcessError(t *testing.T) {
	sf := NewTestSourceFactory()
	done := HandlerFunc(func(blk *pbbstream.Block, obj any) error {
		return fmt.Errorf("please leave")
	})

	mplex := NewMultiplexedSource([]SourceFactory{sf.NewSource}, done)
	go mplex.Run()

	src := <-sf.Created
	<-src.running // test fixture ready to push blocks

	assert.Error(t, src.Push(TestBlock("00000002a", "00000001a"), nil))
	select {
	case <-mplex.Terminating():
	case <-time.After(100 * time.Millisecond):
		t.Errorf("timed out waiting for multiplexedSource to shutdown")
	}

}

func TestMultiplexedSource_noShutdownOnSrcShutdown(t *testing.T) {
	sf := NewTestSourceFactory()
	done := HandlerFunc(func(blk *pbbstream.Block, obj any) error {
		return fmt.Errorf("please leave")
	})

	mplex := NewMultiplexedSource([]SourceFactory{sf.NewSource}, done)
	go mplex.Run()
	src := <-sf.Created

	<-src.running // test fixture ready to push blocks
	src.Shutdown(fmt.Errorf("shutting down source"))
	<-src.Terminating()
	assert.False(t, mplex.IsTerminating(), "multiplexedSource should not go down on source shutdown")

}

func TestMultiplexedSource_retryInterval(t *testing.T) {
	sfOne := NewTestSourceFactory()
	sfTwo := NewTestSourceFactory()
	done := HandlerFunc(func(blk *pbbstream.Block, obj any) error { return nil })

	mplex := NewMultiplexedSource(
		[]SourceFactory{sfOne.NewSource, sfTwo.NewSource},
		done,
		MultiplexedSourceWithRetryIntervals([]time.Duration{0, 200 * time.Millisecond}),
	)
	mplex.reconnectDelay = 10 * time.Millisecond
	go mplex.Run()
	defer mplex.Shutdown(nil)

	srcOne := <-sfOne.Created
	srcTwo := <-sfTwo.Created
	<-srcOne.running
	<-srcTwo.running

	srcOne.Shutdown(fmt.Errorf("test"))
	srcTwo.Shutdown(fmt.Errorf("test"))

	select {
	case <-sfOne.Created:
	case <-time.After(100 * time.Millisecond):
		t.Fatal("source without retry interval was not reconnected on the next pass")
	}

	select {
	case <-sfTwo.Created:
		t.Fatal("source with retry interval was reconnected before its interval elapsed")
	case <-time.After(100 * time.Millisecond):
	}

	select {
	case <-sfTwo.Created:
	case <-time.After(500 * time.Millisecond):
		t.Fatal("source with retry interval was not reconnected after its interval elapsed")
	}
}