network/connection.go

package network

import (
	"git.hpds.cc/Component/network/frame"
	"git.hpds.cc/Component/network/log"
	"git.hpds.cc/Component/network/metadata"
	"io"
	"sync"
)

// Connection wraps the specific io connections (typically quic.Connection) to transfer coder frames
type Connection interface {
	io.Closer

	// Name returns the name of the connection, which is set by clients
	Name() string
	// ClientId connection client ID
	ClientId() string
	// ClientType returns the type of the client (Protocol Gateway | Message Queue | Stream Function)
	ClientType() ClientType
	// Metadata returns the extra info of the application
	Metadata() metadata.Metadata
	// Write should goroutine-safely send coder frames to peer side
	Write(f frame.Frame) error
	// ObserveDataTags observed data tags
	ObserveDataTags() []byte
}

type connection struct {
	name       string
	clientType ClientType
	metadata   metadata.Metadata
	stream     io.ReadWriteCloser
	clientId   string
	observed   []byte // observed data tags
	mu         sync.Mutex
	closed     bool
}

func newConnection(name string, clientId string, clientType ClientType, metadata metadata.Metadata,
	stream io.ReadWriteCloser, observed []byte) Connection {
	return &connection{
		name:       name,
		clientId:   clientId,
		clientType: clientType,
		observed:   observed,
		metadata:   metadata,
		stream:     stream,
		closed:     false,
	}
}

// Close implements io.Close interface
func (c *connection) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.closed = true
	return c.stream.Close()
}

// Name returns the name of the connection, which is set by clients
func (c *connection) Name() string {
	return c.name
}

// ClientType returns the type of the connection (Protocol Gateway | Message Queue | Stream Function )
func (c *connection) ClientType() ClientType {
	return c.clientType
}

// Metadata returns the extra info of the application
func (c *connection) Metadata() metadata.Metadata {
	return c.metadata
}

// Write should goroutine-safely send coder frames to peer side
func (c *connection) Write(f frame.Frame) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		log.Warnf("client stream is closed: %s", c.clientId)
		return nil
	}
	_, err := c.stream.Write(f.Encode())
	return err
}

// ObserveDataTags observed data tags
func (c *connection) ObserveDataTags() []byte {
	return c.observed
}

// ClientId connection client id
func (c *connection) ClientId() string {
	return c.clientId
}