golib/gnet/modbus/buffer_test.go

package modbus

import (
	"context"
	"net"
	"testing"
	"time"

	"golib/gnet"
)

func serverTCPModBus(t *testing.T, address string) {
	ln, err := net.Listen("tcp", address)
	if err != nil {
		t.Error(err)
		return
	}
	ln = gnet.NewListener(ln, &gnet.Config{
		ReadTimout:  5 * time.Second,
		WriteTimout: 2 * time.Second,
	})
	defer func() {
		_ = ln.Close()
	}()
	for {
		conn, err := ln.Accept()
		if err != nil {
			t.Error("serverTCP: accept close:", err)
			return
		}
		go func(conn net.Conn) {
			defer func() {
				_ = conn.Close()
			}()
			for {
				b := make([]byte, gnet.MaxBuffSize)
				n, err := conn.Read(b)
				if err != nil {
					t.Log("conn.Read:", err)
					return
				}

				t.Log("conn.Read:", gnet.Bytes(b[:n]).HexTo())

				p := []byte("hello,world")

				if _, err = conn.Write(p); err != nil {
					t.Log("conn.Write:", err)
					return
				} else {
					t.Log("conn.Write:", string(p))
				}
			}
		}(conn)
	}
}

type mswHandler struct {
	b []byte
}

func (m *mswHandler) Create() ([]byte, error) {
	return m.b, nil
}

func TestNewBuffer(t *testing.T) {
	address := "127.0.0.1:9876"
	go serverTCPModBus(t, address)

	conn, err := gnet.DialTCP("tcp", address)
	if err != nil {
		t.Error(err)
		return
	}

	ctx, cancel := context.WithCancel(context.Background())
	ms := NewBuffer(ctx, conn, &mswHandler{b: []byte(time.Now().String())})
	go ms.Start()
	go func() {
		time.Sleep(5 * time.Second)
		cancel()
	}()

	tk := time.NewTimer(1 * time.Second)
	for {
		select {
		case <-tk.C:
			b, ok := ms.Get()
			if !ok {
				t.Log("Get: continue")
			} else {
				t.Log("client.Read:", string(b))
			}
			tk.Reset(1 * time.Second)
		}
	}
}