rest/application/z_stream_group_test.go

package application

import (
	"context"
	"io"
	"log"
	"testing"
	"time"

	"git.ali33.ru/fcg-xvii/go-tools/json"
	"git.ali33.ru/fcg-xvii/rest"
)

// Пустая реализация IStream
type TestStream struct {
	ctx    context.Context
	cancel context.CancelFunc
}

func NewTestStream() *TestStream {
	ctx, cancel := context.WithCancel(context.Background())
	return &TestStream{
		ctx:    ctx,
		cancel: cancel,
	}
}

func (t *TestStream) SetAuth(auth json.Map) {}

func (t *TestStream) Auth() json.Map { return nil }

func (t *TestStream) SetClientData(key string, val any) {}

func (t *TestStream) ClientData(key string) (any, bool) { return nil, false }

func (t *TestStream) Context() context.Context {
	return t.ctx
}

func (t *TestStream) SendMessage(req rest.IRequestOut) (<-chan *rest.RequestStream, error) {
	log.Printf("TestStream %p send message.\n", t)
	return nil, nil
}

func (t *TestStream) Close() {
	t.cancel()
	log.Printf("TestStream %p closed.\n", t)
}

// Пустая реализация IRequestOut
type TestRequestOut struct {
	TestRequest
}

func NewTestRequestOut() *TestRequestOut {
	return &TestRequestOut{
		TestRequest: TestRequest{},
	}
}

func (r *TestRequestOut) Write(w io.Writer) rest.IErrorArgs {
	// Пустая реализация записи
	return nil
}

// Пустая реализация IRequest
type TestRequest struct{}

func (r *TestRequest) RType() rest.RequestType {
	return 0 // Возвращаем значение по умолчанию
}

func (r *TestRequest) RCommand() string {
	return "" // Возвращаем пустую строку
}

func (r *TestRequest) RData() json.Map {
	return make(json.Map) // Возвращаем пустую карту
}

func (r *TestRequest) RFiles() rest.RequestFiles {
	return nil // Возвращаем nil
}

func (r *TestRequest) RFile(name string) (rest.IReadCloserLen, bool) {
	return nil, false // Возвращаем nil и false
}

func (r *TestRequest) RClose() {
	// Пустая реализация
}

////////////////////////////////////////////////////////

func TestStreamClose(t *testing.T) {
	stream := NewTestStream()
	store := NewStreamStore(stream, context.Background())
	go func() {
		<-store.Context().Done()
		log.Println("store closedddd...")
		store.Close()
		store.Close()
	}()
	stream.Close()
	time.Sleep(time.Second * 5)
}

func TestStreamStoreClose(t *testing.T) {
	stream := NewTestStream()
	store := NewStreamStore(stream, context.Background())
	store.Add(NewTestStream())
	store.Add(NewTestStream())
	store.Add(NewTestStream())
	store.Add(NewTestStream())
	store.Close()
	time.Sleep(time.Minute * 5)
}

func TestGroupClose(t *testing.T) {
	group := NewStreamGroup(context.Background())
	stream := NewTestStream()
	group.Store("1", stream)
	group.Close()
	time.Sleep(time.Minute * 5)
}

func TestGroupStreamClose(t *testing.T) {
	group := NewStreamGroup(context.Background())
	add := func() {
		stream := NewTestStream()
		group.Store("1", stream)
		go func() {
			time.Sleep(time.Second * 3)
			log.Println("remove", stream)
			stream.Close()
		}()
		log.Println("add", stream)
	}
	add()
	add()
	time.Sleep(time.Second * 10)
}

func TestGroupBroadcast(t *testing.T) {
	group := NewStreamGroup(context.Background())
	group.Store("1", NewTestStream())
	group.Store("12", NewTestStream())
	group.EventBroadcast(NewTestRequestOut())
	time.Sleep(time.Second * 10)
}