func (c *BaseConn) SetMaxWriteDelay(delay time.Duration) {
	c.stream.SetMaxWriteDelay(delay)
}
SetMaxWriteDelay will set the maximum amount of time allowed to pass until
an asynchronous write is flushed.
package transport
import (
	"io"
	"sync"
	"time"
	"github.com/256dpi/gomqtt/packet"
)
type Carrier interface {
	io.ReadWriteCloser
	SetReadDeadline(time.Time) error
}
type BaseConn struct {
	carrier      Carrier
	stream       *packet.Stream
	sendMutex    sync.Mutex
	receiveMutex sync.Mutex
	readTimeout  time.Duration
}
func NewBaseConn(c Carrier) *BaseConn {
	return &BaseConn{
		carrier: c,
		stream:  packet.NewStream(c, c),
	}
}
func (c *BaseConn) Send(pkt packet.Generic, async bool) error {
	c.sendMutex.Lock()
	defer c.sendMutex.Unlock()
	err := c.stream.Write(pkt, async)
	if err != nil {
		_ = c.carrier.Close()
		return err
	}
	return nil
}
func (c *BaseConn) Receive() (packet.Generic, error) {
	c.receiveMutex.Lock()
	defer c.receiveMutex.Unlock()
	pkt, err := c.stream.Read()
	if err != nil {
		_ = c.carrier.Close()
		return nil, err
	}
	err = c.resetTimeout()
	if err != nil {
		_ = c.carrier.Close()
		return nil, err
	}
	return pkt, nil
}
func (c *BaseConn) Close() error {
	c.sendMutex.Lock()
	defer c.sendMutex.Unlock()
	err1 := c.stream.Flush()
	err2 := c.carrier.Close()
	if err1 != nil {
		return err1
	} else if err2 != nil {
		return err2
	}
	return nil
}
func (c *BaseConn) SetReadLimit(limit int64) {
	c.stream.SetReadLimit(limit)
}
func (c *BaseConn) SetReadTimeout(timeout time.Duration) {
	c.receiveMutex.Lock()
	defer c.receiveMutex.Unlock()
	c.readTimeout = timeout
	_ = c.resetTimeout()
}