func byteToLength(b byte) int {
	return int(1 << (b + 9))
}
byteToLength returns the value corresponding to a RawSocket lenght byte.
package transport
import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"strings"
	"time"
	"github.com/gammazero/nexus/v3/stdlog"
	"github.com/gammazero/nexus/v3/transport/serialize"
	"github.com/gammazero/nexus/v3/wamp"
)
type rawSocketPeer struct {
	conn       net.Conn
	serializer serialize.Serializer
	sendLimit  int
	recvLimit  int
	closed chan struct{}
	rd chan wamp.Message
	wr chan wamp.Message
	cancelSender context.CancelFunc
	ctxSender    context.Context
	writerDone chan struct{}
	log stdlog.StdLog
}
const (
	rawsocketJSON    = 1
	rawsocketMsgpack = 2
	rawsocketCBOR = 3
	magic = 0x7f
)
func ConnectRawSocketPeer(ctx context.Context, network, addr string, serialization serialize.Serialization, tlsConfig *tls.Config, logger stdlog.StdLog, recvLimit int) (wamp.Peer, error) {
	err := checkNetworkType(network)
	if err != nil {
		return nil, err
	}
	protocol, err := getProtoByte(serialization)
	if err != nil {
		return nil, err
	}
	var d net.Dialer
	conn, err := d.DialContext(ctx, network, addr)
	if err != nil {
		return nil, err
	}
	if tlsConfig != nil {
		colonPos := strings.LastIndex(addr, ":")
		if colonPos == -1 {
			colonPos = len(addr)
		}
		hostname := addr[:colonPos]
		if tlsConfig.ServerName == "" {
			c := tlsConfig.Clone()
			c.ServerName = hostname
			tlsConfig = c
		}
		tlsConn := tls.Client(conn, tlsConfig)
		errChannel := make(chan error, 1)
		go func() {
			errChannel <- tlsConn.Handshake()
		}()
		select {
		case err = <-errChannel:
			if err != nil {
				conn.Close()
				return nil, err
			}
		case <-ctx.Done():
			conn.Close()
			return nil, ctx.Err()
		}
		conn = tlsConn
	}
	peer, err := clientHandshake(conn, logger, protocol, recvLimit)
	if err != nil {
		conn.Close()
		return nil, err
	}
	return peer, nil
}
func AcceptRawSocket(conn net.Conn, logger stdlog.StdLog, recvLimit, outQueueSize int) (wamp.Peer, error) {
	peer, err := serverHandshake(conn, logger, recvLimit, outQueueSize)
	if err != nil {
		conn.Close()
		return nil, err
	}
	return peer, nil
}
func newRawSocketPeer(conn net.Conn, serializer serialize.Serializer, logger stdlog.StdLog, sendLimit, recvLimit, outQueueSize int) *rawSocketPeer {
	rs := &rawSocketPeer{
		conn:       conn,
		serializer: serializer,
		sendLimit:  sendLimit,
		recvLimit:  recvLimit,
		closed:     make(chan struct{}),
		writerDone: make(chan struct{}),
		rd: make(chan wamp.Message),
		wr: make(chan wamp.Message, outQueueSize),
		log: logger,
	}
	rs.ctxSender, rs.cancelSender = context.WithCancel(context.Background())
	go rs.recvHandler()
	go rs.sendHandler()
	return rs
}
func (rs *rawSocketPeer) Recv() <-chan wamp.Message { return rs.rd }
func (rs *rawSocketPeer) TrySend(msg wamp.Message) error {
	return wamp.TrySend(rs.wr, msg)
}
func (rs *rawSocketPeer) SendCtx(ctx context.Context, msg wamp.Message) error {
	return wamp.SendCtx(ctx, rs.wr, msg)
}
func (rs *rawSocketPeer) Send(msg wamp.Message) error {
	return wamp.SendCtx(rs.ctxSender, rs.wr, msg)
}
func (rs *rawSocketPeer) IsLocal() bool { return false }
func (rs *rawSocketPeer) Close() {
	rs.cancelSender()
	<-rs.writerDone
	close(rs.closed)
	rs.conn.Close()
}
func (rs *rawSocketPeer) sendHandler() {
	defer close(rs.writerDone)
	defer rs.cancelSender()
	senderDone := rs.ctxSender.Done()
sendLoop:
	for {
		select {
		case msg := <-rs.wr:
			b, err := rs.serializer.Serialize(msg)
			if err != nil {
				rs.log.Print(err)
				continue sendLoop
			}
			if len(b) > rs.sendLimit {
				rs.log.Println("Message size", len(b), "exceeds limit of",
					rs.sendLimit)
				continue sendLoop
			}
			lenBytes := intToBytes(len(b))
			header := []byte{0x0, lenBytes[0], lenBytes[1], lenBytes[2]}
			if _, err = rs.conn.Write(header); err != nil {
				if !wamp.IsGoodbyeAck(msg) {
					rs.log.Println("Error writing header:", err)
				}
				continue sendLoop
			}
			if _, err = rs.conn.Write(b); err != nil {
				if !wamp.IsGoodbyeAck(msg) {
					rs.log.Println("Error writing message:", msg, err)
				}
				continue sendLoop
			}
		case <-senderDone:
			return
		}
	}
}
func (rs *rawSocketPeer) recvHandler() {
	defer close(rs.rd)
MsgLoop:
	for {
		var header [4]byte
		_, err := io.ReadFull(rs.conn, header[:])
		if err != nil {
			select {
			case <-rs.closed:
			default:
				rs.cancelSender()
				<-rs.writerDone
				rs.conn.Close()
			}
			return
		}
		length := bytesToInt(header[1:])
		if length > rs.recvLimit {
			rs.log.Print("Received message that exceeded size limit, closing")
			rs.conn.Close()
			break
		}
		var msg wamp.Message
		switch header[0] & 0x07 {
		case 0:  
			buf := make([]byte, length)
			_, err = io.ReadFull(rs.conn, buf)
			if err != nil {
				rs.log.Println("Error reading message:", err)
				rs.conn.Close()
				return
			}
			msg, err = rs.serializer.Deserialize(buf)
			if err != nil {
				rs.log.Println("Cannot deserialize peer message:", err)
				continue MsgLoop
			}
		case 1:  
			header[0] = 0x02
			if _, err = rs.conn.Write(header[:]); err != nil {
				rs.log.Println("Error writing header responding to PING:", err)
				rs.conn.Close()
				return
			}
			if _, err = io.CopyN(rs.conn, rs.conn, int64(length)); err != nil {
				rs.log.Println("Error responding to PING:", err)
				rs.conn.Close()
				return
			}
			continue MsgLoop
		case 2:  
			_, err = io.CopyN(ioutil.Discard, rs.conn, int64(length))
			if err != nil {
				rs.log.Println("Error reading PONG:", err)
				rs.conn.Close()
				return
			}
			continue MsgLoop
		}
		select {
		case rs.rd <- msg:
		case <-rs.closed:
			select {
			case rs.rd <- msg:
			case <-time.After(time.Second):
				rs.conn.Close()
				return
			}
		}
	}
}
func clientHandshake(conn net.Conn, logger stdlog.StdLog, protocol byte, recvLimit int) (*rawSocketPeer, error) {
	maxRecvLen := fitRecvLimit(recvLimit)
	_, err := conn.Write([]byte{magic, (maxRecvLen&0xf)<<4 | protocol, 0, 0})
	if err != nil {
		return nil, fmt.Errorf("error sending handshake: %s", err)
	}
	var buf [4]byte
	if _, err = io.ReadFull(conn, buf[:]); err != nil {
		return nil, err
	}
	if buf[0] != magic {
		return nil, errors.New("not a rawsocket handshake")
	}
	repSerializer := buf[1] & 0xf
	if repSerializer == 0 {
		errCode := buf[1] >> 4
		switch errCode {
		case 0:
			return nil, errors.New("illegal error code")
		case 1:
			return nil, errors.New("serializer unsupported")
		case 2:
			return nil, errors.New("maximum message length unacceptable")
		case 3:
			return nil, errors.New("use of reserved bits (unsupported feature)")
		case 4:
			return nil, errors.New("maximum connection count reached")
		default:
			return nil, fmt.Errorf("unknown error: %d", errCode)
		}
	}
	if repSerializer != protocol {
		return nil, errors.New("serializer mismatch")
	}
	var serializer serialize.Serializer
	switch protocol {
	case rawsocketJSON:
		serializer = &serialize.JSONSerializer{}
	case rawsocketMsgpack:
		serializer = &serialize.MessagePackSerializer{}
	case rawsocketCBOR:
		serializer = &serialize.CBORSerializer{}
	}
	sendLimit := byteToLength(buf[1] >> 4)
	recvLimit = byteToLength(maxRecvLen)
	return newRawSocketPeer(conn, serializer, logger, sendLimit, recvLimit, 0), nil
}
func serverHandshake(conn net.Conn, logger stdlog.StdLog, recvLimit, outQueueSize int) (*rawSocketPeer, error) {
	var buf [4]byte
	if _, err := io.ReadFull(conn, buf[:]); err != nil {
		return nil, err
	}
	if buf[0] != magic {
		return nil, errors.New("not a rawsocket handshake")
	}
	if buf[2] != 0 || buf[3] != 0 {
		conn.Write([]byte{magic, byte(0x3 << 4), 0, 0})
		return nil, errors.New("use of reserved bits (unsupported feature)")
	}
	serialization := buf[1] & 0xf
	var serializer serialize.Serializer
	switch serialization {
	case 0:
		return nil, errors.New("illegal serializer value")
	case rawsocketJSON:
		serializer = &serialize.JSONSerializer{}
	case rawsocketMsgpack:
		serializer = &serialize.MessagePackSerializer{}
	case rawsocketCBOR:
		serializer = &serialize.CBORSerializer{}
	default:
		conn.Write([]byte{magic, byte(0x1 << 4), 0, 0})
		return nil, errors.New("serializer unsupported")
	}
	maxRecvLen := fitRecvLimit(recvLimit)
	_, err := conn.Write([]byte{magic, maxRecvLen<<4 | serialization, 0, 0})
	if err != nil {
		return nil, fmt.Errorf("error sending handshake: %s", err)
	}
	sendLimit := byteToLength(buf[1] >> 4)
	recvLimit = byteToLength(maxRecvLen)
	return newRawSocketPeer(conn, serializer, logger, sendLimit, recvLimit, outQueueSize), nil
}
func fitRecvLimit(recvLimit int) byte {
	if recvLimit > 0 {
		for b := byte(0); b < 0xf; b++ {
			if byteToLength(b) >= recvLimit {
				return b
			}
		}
	}
	return 0xf
}
func intToBytes(i int) [3]byte {
	return [3]byte{
		byte((i >> 16) & 0xff),
		byte((i >> 8) & 0xff),
		byte(i & 0xff),
	}
}
func bytesToInt(b []byte) int {
	var n, shift uint
	for i := len(b) - 1; i >= 0; i-- {
		n |= uint(b[i]) << shift
		shift += 8
	}
	return int(n)
}