SVI Доблавение кодов, тестов; 62.2%

internal/dict_topic/topic/topic_test.go

@@ -86,7 +86,7 @@ func (sf *tester) subGood1() {
 	sf.t.Log("subGood1")
 	req := &netapi.BuffRequest{
 		Sample:             "",
-		ClientName:         "",
+		ClientName:         "test_client",
 		MsgSumSizeLimit:    0,
 		MsgSaveSecondLimit: 0,
 	}

internal/serv_grpc/serv_subcsribe/serv_subscribe.go

@@ -6,7 +6,7 @@ import (
 	"log"
 
 	"p78git.ddns.net/svi/gobus/api/netapi"
-	"p78git.ddns.net/svi/gobus/pkg/net/cleint_proxy"
+	"p78git.ddns.net/svi/gobus/pkg/net/client_proxy"
 	"p78git.ddns.net/svi/gobus/pkg/types"
 )
 
@@ -31,7 +31,7 @@ func NewServSubscribe(serv types.IService) (*ServSubscribe, error) {
 
 // Subscribe -- подписывает на топики по шаблону
 func (sf *ServSubscribe) Subscribe(req *netapi.SubscribeRequest, serv netapi.GoBus_SubscribeServer) error {
-	clientProxy, err := cleint_proxy.NewClientProxy(req)
+	clientProxy, err := client_proxy.NewClientProxy(req)
 	if err != nil {
 		return fmt.Errorf("ServSubscribe.Subscribe(): in create ClientProxy, err=\n\t%w", err)
 	}

pkg/net/cleint_proxy/client_proxy.go → pkg/net/client_proxy/client_proxy.go

@@ -1,5 +1,5 @@
-// package cleint_proxy -- прокси подписывания клиента на топики по шаблону
-package cleint_proxy
+// package client_proxy -- прокси подписывания клиента на топики по шаблону
+package client_proxy
 
 import (
 	"fmt"

pkg/net/cleint_proxy/client_proxy_test.go → pkg/net/client_proxy/client_proxy_test.go

@@ -1,4 +1,4 @@
-package cleint_proxy
+package client_proxy
 
 import (
 	"strings"

pkg/net/client_proxy_buffer/client_proxy_buffer.go

@@ -19,29 +19,37 @@ type ClientProxyBuffer struct {
 	name            alias.ClientName      // Имя клиента
 	sample          alias.TopicName       // Образец подписи
 	isWork          bool                  // Признак, что получатель живой
+	isLost          bool                  // Признак, что потеряны данные
 	lstMsg          []*netapi.TopicMsg    // Список очереди ожидания сообщений клиента
 	sizeByteLimit   int                   // Предельный размер буфера в байтах
 	sizeByteCurrent int                   // Текущий размер буфера в байтах
 	chMsg           chan *netapi.TopicMsg // Канал для обратной связи из топиков
+	chBuff          chan *netapi.TopicMsg // Канал временного буфера
 	block           sync.RWMutex
 }
 
 // NewClientProxyBuffer -- возвращает новый буферизованный прокси клиента
 func NewClientProxyBuffer(req *netapi.BuffRequest) (*ClientProxyBuffer, error) {
-	if req == nil {
-		return nil, fmt.Errorf("NewClientProxyBuffer(): BuffRequest==nil")
+	{ // Предусловия
+		if req == nil {
+			return nil, fmt.Errorf("NewClientProxyBuffer(): SubscribeRequest==nil")
+		}
+		if req.ClientName == "" {
+			return nil, fmt.Errorf("NewClientProxyBuffer(): name is empty")
+		}
 	}
-	log.Printf("NewClientProxy(): name=%q, sample=%q\n", req.ClientName, req.Sample)
-	suffix := "_" + alias.ClientName(fmt.Sprint(time.Now().UnixMilli()))
+	log.Printf("NewClientProxyBuffer(): name=%q, sample=%q\n", req.ClientName, req.Sample)
 	sf := &ClientProxyBuffer{
-		name:          alias.ClientName(req.ClientName) + suffix,
+		name:          alias.ClientName(req.ClientName),
 		sample:        alias.TopicName(req.Sample),
 		chMsg:         make(chan *netapi.TopicMsg, 10),
+		chBuff:        make(chan *netapi.TopicMsg, 10),
 		lstMsg:        make([]*netapi.TopicMsg, 0),
 		sizeByteLimit: int(req.MsgSumSizeLimit),
 		isWork:        true,
 	}
-	_ = types.IClientProxy(sf)
+	go sf.workBuff()
+	_ = types.IClientProxyBuffer(sf)
 	return sf, nil
 }
 
@@ -73,7 +81,17 @@ func (sf *ClientProxyBuffer) Write(msg *netapi.TopicMsg) {
 	if !sf.isWork {
 		return
 	}
-	sf.chMsg <- msg
+	if msg == nil {
+		log.Printf("ClientProxy.Write(): clientName=%q, msg==nil\n", sf.name)
+		return
+	}
+	// Проверка буфера
+	if len(sf.chMsg) < 10 {
+		sf.chMsg <- msg
+		return
+	}
+	// Буфер заполнен. Добавить в очередь
+	sf.chBuff <- msg
 }
 
 // SampleTopic -- возвращает образец шаблона подписки на топики
@@ -85,3 +103,44 @@ func (sf *ClientProxyBuffer) SampleTopic() alias.TopicName {
 func (sf *ClientProxyBuffer) ClientName() alias.ClientName {
 	return sf.name
 }
+
+// Работает в отдельном потоке, добавляет в буфер собщения ,если канал отдачи клиенту забит.
+func (sf *ClientProxyBuffer) workBuff() {
+	fnSend := func() string { // Проверка с блокировкой доступности канала отправки
+		sf.block.Lock()
+		defer sf.block.Unlock()
+		if len(sf.chMsg) >= 10 {
+			return "full"
+		}
+		if len(sf.lstMsg) == 0 {
+			sf.sizeByteCurrent = 0
+			return "empty"
+		}
+		msg := sf.lstMsg[0]
+		sf.sizeByteCurrent -= len(msg.BinMsg) + len(msg.Topic)
+		sf.chMsg <- msg
+		sf.lstMsg = sf.lstMsg[:1]
+		return "send"
+	}
+	for sf.IsWork() { // Цикл добавления сообщений в буфер и вычитки из него
+		select {
+		case msg := <-sf.chBuff: // Сообщение переполнения
+			if sf.sizeByteCurrent >= sf.sizeByteLimit {
+				sf.isLost = true
+				continue
+			}
+			sf.sizeByteCurrent += len(msg.BinMsg) + len(msg.Topic)
+			sf.lstMsg = append(sf.lstMsg, msg)
+			continue
+		default:
+			switch res := fnSend(); res {
+			case "full": // чередь чтения забита, немного подождать
+				time.Sleep(time.Millisecond * 1000)
+			case "empty": // Очередь пустая долго спим
+				time.Sleep(time.Millisecond * 5000)
+			case "send": // Что-то отправили, сразу проверить повторную отправку
+				continue
+			}
+		}
+	}
+}

pkg/net/client_proxy_buffer/client_proxy_buffer_test.go

@@ -1,13 +1,19 @@
 package client_proxy_buffer
 
-import "testing"
+import (
+	"strings"
+	"testing"
+
+	"p78git.ddns.net/svi/gobus/api/netapi"
+)
 
 /*
-	Тест для ...
+	Тест для прокси клиента
 */
 
 type tester struct {
-	t *testing.T
+	t     *testing.T
+	proxy *ClientProxyBuffer
 }
 
 func TestAny(t *testing.T) {
@@ -15,8 +21,99 @@ func TestAny(t *testing.T) {
 		t: t,
 	}
 	sf.create()
+	sf.write()
+}
+
+// Запись сообщения в прокси клиента
+func (sf *tester) write() {
+	sf.t.Log("write")
+	sf.writeBad1()
+	sf.writeGood1()
+	sf.read()
+}
+
+// Читает сообщение из шины
+func (sf *tester) read() {
+	sf.t.Log("read")
+	msg := sf.proxy.Read()
+	if msg == nil {
+		sf.t.Fatalf("read(): msg==nil")
+	}
+	sf.proxy.ResetWork()
+	sf.writeGood1()
+}
+
+func (sf *tester) writeGood1() {
+	sf.t.Log("writeGood1")
+	msg := &netapi.TopicMsg{
+		Source:  0,
+		Topic:   "",
+		BinMsg:  []byte{},
+		StrUuid: "",
+	}
+	sf.proxy.Write(msg)
+}
+
+// Нет сообщения для прокси
+func (sf *tester) writeBad1() {
+	sf.t.Log("writeBad1")
+	sf.proxy.Write(nil)
 }
 
+// Создание прокси для клиента
 func (sf *tester) create() {
 	sf.t.Log("create")
+	sf.createBad1()
+	sf.createBad2()
+	sf.createGood1()
+}
+
+func (sf *tester) createGood1() {
+	sf.t.Log("createGood1")
+	req := &netapi.BuffRequest{
+		Sample:     "_",
+		ClientName: "test_client",
+	}
+	var err error
+	sf.proxy, err = NewClientProxyBuffer(req)
+	if err != nil {
+		sf.t.Fatalf("createGood1(): err=%v", err)
+	}
+	if sf.proxy == nil {
+		sf.t.Fatalf("createGood1(): proxy==nil")
+	}
+	if !sf.proxy.IsWork() {
+		sf.t.Fatalf("createGood1(): not work")
+	}
+	if sample := sf.proxy.SampleTopic(); sample != "_" {
+		sf.t.Fatalf("createGood1(): sample(%q)!='_'", sample)
+	}
+	if !strings.HasPrefix(string(sf.proxy.ClientName()), "test_client") {
+		sf.t.Fatalf("createGood1(): not have prefix 'test_client', clientName(%q)", sf.proxy.ClientName())
+	}
+}
+
+// В запросе нет имени клиента
+func (sf *tester) createBad2() {
+	sf.t.Log("createBad2")
+	req := &netapi.BuffRequest{}
+	proxy, err := NewClientProxyBuffer(req)
+	if err == nil {
+		sf.t.Fatalf("createBad2(): err==nil")
+	}
+	if proxy != nil {
+		sf.t.Fatalf("createBad2(): proxy!=nil")
+	}
+}
+
+// Нет запроса для создания прокси клиента
+func (sf *tester) createBad1() {
+	sf.t.Log("createBad1")
+	proxy, err := NewClientProxyBuffer(nil)
+	if err == nil {
+		sf.t.Fatalf("createBad1(): err==nil")
+	}
+	if proxy != nil {
+		sf.t.Fatalf("createBad1(): proxy!=nil")
+	}
 }

pkg/net/client_proxy_buffer/proxy_buff/proxy_buff.go

@@ -0,0 +1,2 @@
+// package proxy_buff -- прокси буфер для пуферизованного прокси клиента
+package proxy_buff