bus_test.go

// This file is part of the Smart Home
// Program complex distribution https://github.com/e154/smart-home
// Copyright (C) 2023, Filippov Alex
//
// This library is free software: you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 3 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Library General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library.  If not, see
// <https://www.gnu.org/licenses/>.

package bus

import (
	"context"
	"fmt"
	"reflect"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"go.uber.org/atomic"
)

func TestBus(t *testing.T) {

	const topic = "test/topic"

	b := NewBus()

	var counter = 0
	var wg = sync.WaitGroup{}

	// Test Subscribe
	fn := func(topic string, arg1 string, arg2 string) {
		counter++
		wg.Done()
	}
	wg.Add(1)
	err := b.Subscribe(topic, fn)
	if err != nil {
		t.Errorf("Subscribe returned an error: %v", err)
	}

	// Test Publish
	b.Publish(topic, "hello", "world")

	wg.Wait()

	require.Equal(t, counter, 1)

	// ------------------------------------------------------------

	// Test Stat
	stats, total, err := b.Stat(context.Background(), 999, 0, "", "")
	if err != nil {
		t.Errorf("Stat returned an error: %v", err)
	}
	if total != 1 {
		t.Errorf("Stat returned a non-zero total: %d", total)
	}
	if len(stats) != 1 {
		t.Errorf("Stat returned a non-empty stats slice: %v", stats)
	}

	require.Equal(t, stats[0].Topic, topic)
	require.Equal(t, stats[0].Subscribers, 1)

	// ------------------------------------------------------------

	// Test Unsubscribe
	err = b.Unsubscribe(topic, fn)
	if err != nil {
		t.Errorf("Unsubscribe returned an error: %v", err)
	}

	// Test Publish
	b.Publish(topic, "hello", "world")

	time.Sleep(time.Second)

	require.Equal(t, counter, 1)

	// ------------------------------------------------------------

	// Test Stat
	stats, total, err = b.Stat(context.Background(), 999, 0, "", "")
	if err != nil {
		t.Errorf("Stat returned an error: %v", err)
	}
	if total != 0 {
		t.Errorf("Stat returned a non-zero total: %d", total)
	}
	if len(stats) != 0 {
		t.Errorf("Stat returned a non-empty stats slice: %v", stats)
	}

	// ------------------------------------------------------------

	// Test Subscribe
	fn = func(topic string, arg1 string, arg2 string) {
		counter++
	}
	err = b.Subscribe(topic, fn)
	if err != nil {
		t.Errorf("Subscribe returned an error: %v", err)
	}

	// Test Close
	b.CloseTopic(topic)

	// Test Publish
	b.Publish(topic, "hello", "world")

	time.Sleep(time.Second)

	require.Equal(t, 1, counter)

	// Test Stat
	stats, total, err = b.Stat(context.Background(), 999, 0, "", "")
	if err != nil {
		t.Errorf("Stat returned an error: %v", err)
	}
	if total != 0 {
		t.Errorf("Stat returned a non-zero total: %d", total)
	}
	if len(stats) != 0 {
		t.Errorf("Stat returned a non-empty stats slice: %v", stats)
	}

	// ------------------------------------------------------------

	fn = func(topic string, arg1 string, arg2 string) {
		counter++
	}
	err = b.Subscribe(topic, fn)
	if err != nil {
		t.Errorf("Subscribe returned an error: %v", err)
	}

	// Test Close
	b.Purge()

	// Test Publish
	b.Publish(topic, "hello", "world")

	time.Sleep(time.Second)

	require.Equal(t, counter, 1)

	// Test Stat
	stats, total, err = b.Stat(context.Background(), 999, 0, "", "")
	if err != nil {
		t.Errorf("Stat returned an error: %v", err)
	}
	if total != 0 {
		t.Errorf("Stat returned a non-zero total: %d", total)
	}
	if len(stats) != 0 {
		t.Errorf("Stat returned a non-empty stats slice: %v", stats)
	}

	// ------------------------------------------------------------

	// Test buildHandlerArgs
	args := buildHandlerArgs([]interface{}{topic, "hello", "world"})
	if len(args) != 3 {
		t.Errorf("buildHandlerArgs returned the wrong number of arguments: %v", args)
	}
	if args[0].String() != topic {
		t.Errorf("buildHandlerArgs returned the wrong topic: %v", args[0])
	}
	if args[1].String() != "hello" {
		t.Errorf("buildHandlerArgs returned the wrong arg1: %v", args[1])
	}
	if args[2].String() != "world" {
		t.Errorf("buildHandlerArgs returned the wrong arg2: %v", args[2])
	}

	// Test reflection of buildHandlerArgs
	if reflect.TypeOf(buildHandlerArgs).Kind() != reflect.Func {
		t.Errorf("buildHandlerArgs is not a function")
	}
}

func TestBus2(t *testing.T) {

	const topic = "test/topic"

	b := NewBus()

	var counter atomic.Int32
	var wg = sync.WaitGroup{}

	// Test Subscribe
	fn := func(topic string, arg1 string, arg2 string) {
		fmt.Println("fn1")
		counter.Inc()
		wg.Done()
	}

	fn2 := func(topic string, arg1 string, arg2 string) {
		fmt.Println("fn2")
		counter.Inc()
		wg.Done()
	}

	fn3 := func(topic string, arg1 string, arg2 string) {
		fmt.Println("fn3")
		counter.Inc()
		wg.Done()
	}

	wg.Add(3)

	err := b.Subscribe(topic, fn)
	if err != nil {
		t.Errorf("Subscribe returned an error: %v", err)
	}
	err = b.Subscribe(topic, fn2)
	if err != nil {
		t.Errorf("Subscribe returned an error: %v", err)
	}
	err = b.Subscribe(topic, fn3)
	if err != nil {
		t.Errorf("Subscribe returned an error: %v", err)
	}

	// Test Stat
	stats, total, err := b.Stat(context.Background(), 999, 0, "", "")
	if err != nil {
		t.Errorf("Stat returned an error: %v", err)
	}
	if total != 1 {
		t.Errorf("Stat returned a non-zero total: %d", total)
	}
	if len(stats) != 1 {
		t.Errorf("Stat returned a non-empty stats slice: %v", stats)
	}

	// Test Publish
	b.Publish(topic, "hello", "world")

	wg.Wait()

	require.Equal(t, int32(3), counter.Load())
}

func TestBus3(t *testing.T) {

	bus := NewBus()

	var counter atomic.Int32
	var wg = sync.WaitGroup{}

	const n = 1000

	wg.Add(n)
	fn := func(_ string, msg interface{}) {
		//fmt.Println("msg", msg)
		counter.Inc()
		wg.Done()
	}

	for i := 0; i < n; i++ {
		_ = bus.Subscribe(fmt.Sprintf("foo/bar/%d", i), fn)
	}

L:
	stat, total, err := bus.Stat(context.Background(), 1000, 0, "", "")
	require.NoError(t, err)

	if total < 1000 {
		goto L
	}

	require.NoError(t, err)
	require.Equal(t, len(stat), n)
	require.Equal(t, total, int64(n))
	require.Equal(t, counter.Load(), int32(0))

	counter.Store(0)

	for i := 0; i < n; i++ {
		bus.Publish(fmt.Sprintf("foo/bar/%d", i), i)
	}

L2:
	if counter.Load() < 1000 {
		time.Sleep(time.Second)
		goto L2
	}

	for i := 0; i < n; i++ {
		_ = bus.Unsubscribe(fmt.Sprintf("foo/bar/%d", i), fn)
	}

L3:
	stat, total, err = bus.Stat(context.Background(), 1000, 0, "", "")
	require.NoError(t, err)

	if total != 0 {
		goto L3
	}

	require.NoError(t, err)
	require.Equal(t, len(stat), 0)
	require.Equal(t, total, int64(0))
	require.Equal(t, counter.Load(), int32(n))

	counter.Store(0)

	for i := 0; i < n; i++ {
		bus.Publish(fmt.Sprintf("foo/bar/%d", i), i)
	}

	time.Sleep(time.Second * 5)

	require.True(t, counter.Load() == 0)
}

func TestBus4(t *testing.T) {

	bus := NewBus()

	var counter1 atomic.Int32
	var counter2 atomic.Int32
	var wg1 = sync.WaitGroup{}
	var wg2 = sync.WaitGroup{}

	const n = 1

	wg1.Add(n)
	fn1 := func(_ string, msg interface{}) {
		//fmt.Println("msg", msg)
		counter1.Inc()
		wg1.Done()
	}
	wg2.Add(n)
	fn2 := func(_ string, msg interface{}) {
		//fmt.Println("msg", msg)
		counter2.Inc()
		wg2.Done()
	}

	for i := 0; i < n; i++ {
		_ = bus.Subscribe(fmt.Sprintf("foo/bar/%d", i), fn1)
		_ = bus.Subscribe(fmt.Sprintf("foo/bar/%d", i), fn2)
	}

	time.Sleep(time.Second)

	stat, total, err := bus.Stat(context.Background(), 999, 0, "", "")
	require.NoError(t, err)
	require.Equal(t, len(stat), n)
	require.Equal(t, total, int64(n))
	require.Equal(t, counter1.Load(), int32(0))
	require.Equal(t, counter2.Load(), int32(0))

	for i := 0; i < n; i++ {
		bus.Publish(fmt.Sprintf("foo/bar/%d", i), i)
	}

	wg1.Wait()
	wg2.Wait()

	require.Equal(t, counter1.Load(), int32(n))
	require.Equal(t, counter2.Load(), int32(n))

	for i := 0; i < n; i++ {
		_ = bus.Unsubscribe(fmt.Sprintf("foo/bar/%d", i), fn1)
	}
	time.Sleep(time.Second)

	stat, total, err = bus.Stat(context.Background(), 999, 0, "", "")
	require.NoError(t, err)
	require.Equal(t, len(stat), n)
	require.Equal(t, total, int64(n))

	wg2.Add(n)
	for i := 0; i < n; i++ {
		bus.Publish(fmt.Sprintf("foo/bar/%d", i), i)
	}

	wg2.Wait()

	require.Equal(t, counter1.Load(), int32(n))
	require.Equal(t, counter2.Load(), int32(n*2))

	stat, total, err = bus.Stat(context.Background(), 999, 0, "", "")
	require.NoError(t, err)
	require.Equal(t, len(stat), n)
	require.Equal(t, total, int64(n))
}

func BenchmarkBus(b *testing.B) {

	const topic = "test/topic"

	bus := NewBus()

	var counter atomic.Int32

	// Test Subscribe
	fn := func(topic string, arg1 string, arg2 string) {
		counter.Inc()
	}
	err := bus.Subscribe(topic, fn)
	require.NoError(b, err)

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		bus.Publish(topic, "hello", "world")
	}

	time.Sleep(time.Second)

	require.Equal(b, int32(b.N), counter.Load())
}