func (c *Async) WriteBufferSize() int {
	c.Lock()
	if c.state.Load() != CONNECTED {
		c.Unlock()
		return 0
	}
	i := c.writer.Buffered()
	c.Unlock()
	return i
}
WriteBufferSize returns the size of the underlying message buffer (used for internal message handling and for heartbeat logic)
package frisbee
import (
	"bufio"
	"bytes"
	"crypto/tls"
	"encoding/binary"
	"github.com/loopholelabs/frisbee/internal/errors"
	"github.com/loopholelabs/frisbee/internal/protocol"
	"github.com/loopholelabs/frisbee/internal/ringbuffer"
	"github.com/rs/zerolog"
	"go.uber.org/atomic"
	"io"
	"net"
	"os"
	"sync"
	"time"
)
type Async struct {
	sync.Mutex
	conn             net.Conn
	state            *atomic.Int32
	writer           *bufio.Writer
	flusher          chan struct{}
	incomingMessages *ringbuffer.RingBuffer
	streamConnMutex  sync.RWMutex
	streamConns      map[uint64]*Stream
	streamConnCh     chan *Stream
	logger           *zerolog.Logger
	wg               sync.WaitGroup
	error            *atomic.Error
}
func ConnectAsync(addr string, keepAlive time.Duration, logger *zerolog.Logger, TLSConfig *tls.Config) (*Async, error) {
	var conn net.Conn
	var err error
	if TLSConfig != nil {
		conn, err = tls.Dial("tcp", addr, TLSConfig)
	} else {
		conn, err = net.Dial("tcp", addr)
		_ = conn.(*net.TCPConn).SetKeepAlive(true)
		_ = conn.(*net.TCPConn).SetKeepAlivePeriod(keepAlive)
	}
	if err != nil {
		return nil, errors.WithContext(err, DIAL)
	}
	return NewAsync(conn, logger), nil
}
func NewAsync(c net.Conn, logger *zerolog.Logger) (conn *Async) {
	conn = &Async{
		conn:             c,
		state:            atomic.NewInt32(CONNECTED),
		writer:           bufio.NewWriterSize(c, DefaultBufferSize),
		incomingMessages: ringbuffer.NewRingBuffer(DefaultBufferSize),
		streamConns:      make(map[uint64]*Stream),
		streamConnCh:     make(chan *Stream, 1024),
		flusher:          make(chan struct{}, 1024),
		logger:           logger,
		error:            atomic.NewError(nil),
	}
	if logger == nil {
		conn.logger = &defaultLogger
	}
	conn.wg.Add(2)
	go conn.flushLoop()
	go conn.readLoop()
	return
}
func (c *Async) SetDeadline(t time.Time) error {
	return c.conn.SetDeadline(t)
}
func (c *Async) SetReadDeadline(t time.Time) error {
	return c.conn.SetReadDeadline(t)
}
func (c *Async) SetWriteDeadline(t time.Time) error {
	return c.conn.SetWriteDeadline(t)
}
func (c *Async) ConnectionState() (tls.ConnectionState, error) {
	if tlsConn, ok := c.conn.(*tls.Conn); ok {
		return tlsConn.ConnectionState(), nil
	}
	return tls.ConnectionState{}, NotTLSConnectionError
}
func (c *Async) LocalAddr() net.Addr {
	return c.conn.LocalAddr()
}
func (c *Async) RemoteAddr() net.Addr {
	return c.conn.RemoteAddr()
}
func (c *Async) StreamChannel() <-chan *Stream {
	return c.streamConnCh
}
func (c *Async) WriteMessage(message *Message, content *[]byte) error {
	if content != nil && int(message.ContentLength) != len(*content) {
		return InvalidContentLength
	}
	encodedMessage := [protocol.MessageSize]byte{protocol.ReservedBytes[0], protocol.ReservedBytes[1], protocol.ReservedBytes[2], protocol.ReservedBytes[3]}
	binary.BigEndian.PutUint32(encodedMessage[protocol.FromOffset:protocol.FromOffset+protocol.FromSize], message.From)
	binary.BigEndian.PutUint32(encodedMessage[protocol.ToOffset:protocol.ToOffset+protocol.ToSize], message.To)
	binary.BigEndian.PutUint64(encodedMessage[protocol.IdOffset:protocol.IdOffset+protocol.IdSize], message.Id)
	binary.BigEndian.PutUint32(encodedMessage[protocol.OperationOffset:protocol.OperationOffset+protocol.OperationSize], message.Operation)
	binary.BigEndian.PutUint64(encodedMessage[protocol.ContentLengthOffset:protocol.ContentLengthOffset+protocol.ContentLengthSize], message.ContentLength)
	c.Lock()
	if c.state.Load() != CONNECTED {
		c.Unlock()
		return c.Error()
	}
	_, err := c.writer.Write(encodedMessage[:])
	if err != nil {
		c.Unlock()
		if c.state.Load() != CONNECTED {
			err = c.Error()
			c.logger.Error().Msgf(errors.WithContext(err, WRITE).Error())
			return errors.WithContext(err, WRITE)
		}
		c.logger.Error().Msgf(errors.WithContext(err, WRITE).Error())
		return c.closeWithError(err)
	}
	if content != nil {
		_, err = c.writer.Write(*content)
		if err != nil {
			c.Unlock()
			if c.state.Load() != CONNECTED {
				err = c.Error()
				c.logger.Error().Msgf(errors.WithContext(err, WRITE).Error())
				return errors.WithContext(err, WRITE)
			}
			c.logger.Error().Msgf(errors.WithContext(err, WRITE).Error())
			return c.closeWithError(err)
		}
	}
	if len(c.flusher) == 0 {
		select {
		case c.flusher <- struct{}{}:
		default:
		}
	}
	c.Unlock()
	return nil
}
func (c *Async) ReadMessage() (*Message, *[]byte, error) {
	if c.state.Load() != CONNECTED {
		return nil, nil, c.Error()
	}
	readPacket, err := c.incomingMessages.Pop()
	if err != nil {
		if c.state.Load() != CONNECTED {
			err = c.Error()
			c.logger.Error().Msgf(errors.WithContext(err, POP).Error())
			return nil, nil, errors.WithContext(err, POP)
		}
		c.logger.Error().Msgf(errors.WithContext(err, POP).Error())
		return nil, nil, errors.WithContext(c.closeWithError(err), POP)
	}
	return (*Message)(readPacket.Message), readPacket.Content, nil
}
func (c *Async) Flush() error {
	c.Lock()
	if c.writer.Buffered() > 0 {
		err := c.SetWriteDeadline(time.Now().Add(defaultDeadline))
		if err != nil {
			c.Unlock()
			return c.closeWithError(err)
		}
		err = c.writer.Flush()
		_ = c.SetWriteDeadline(time.Time{})
		if err != nil {
			c.Unlock()
			return c.closeWithError(err)
		}
	}
	c.Unlock()
	return nil
}