<commit_before>package att

import (
	"bytes"
	"encoding/binary"
	"io"
	"log"
	"time"

	"github.com/currantlabs/bt"
	"github.com/currantlabs/bt/uuid"
)

// Server implementas an ATT (Attribute Protocol) server.
type Server struct {
	l2c   bt.Conn
	attrs *Range

	// Refer to [Vol 3, Part F, 3.3.2 & 3.3.3] for the requirement of
	// sequential request-response protocol, and transactions.
	txBuf     []byte
	rxBuf     []byte
	chNotBuf  chan []byte
	chIndBuf  chan []byte
	chConfirm chan bool
}

// NewServer returns an ATT (Attribute Protocol) server.
func NewServer(a *Range, l2c bt.Conn, rxMTU int) *Server {
	// Although the rxBuf is initialized with the capacity of rxMTU, it is
	// not discovered, and only the default ATT_MTU (23 bytes) of it shall
	// be used until remote central request ExchangeMTU.
	s := &Server{
		l2c:   l2c,
		attrs: a,

		txBuf:     make([]byte, 23, 23),
		rxBuf:     make([]byte, rxMTU, rxMTU),
		chNotBuf:  make(chan []byte, 1),
		chIndBuf:  make(chan []byte, 1),
		chConfirm: make(chan bool, 1),
	}
	s.chNotBuf <- make([]byte, 23, 23)
	s.chIndBuf <- make([]byte, 23, 23)
	return s
}

// Notify sends notification to remote central.
func (s *Server) Notify(ind bool, h uint16, data []byte) (int, error) {
	if ind {
		return s.indicate(h, data)
	}
	return s.notify(h, data)
}

// notify sends notification to remote central.
func (s *Server) notify(h uint16, data []byte) (int, error) {
	// log.Printf("att: notifying 0x%04X, %s", h, string(data))

	// Acquire and reuse notifyBuffer. Release it after usage.
	nBuf := <-s.chNotBuf
	defer func() { s.chNotBuf <- nBuf }()

	rsp := HandleValueNotification(nBuf)
	rsp.SetAttributeOpcode()
	rsp.SetAttributeHandle(h)
	buf := bytes.NewBuffer(rsp.AttributeValue())
	buf.Reset()
	if len(data) > buf.Cap() {
		data = data[:buf.Cap()]
	}
	buf.Write(data)
	return s.l2c.Write(rsp[:3+buf.Len()])
}

// indicate sends indication to remote central.
func (s *Server) indicate(h uint16, data []byte) (int, error) {
	// log.Printf("att: indicating 0x%04X, %s", h, string(data))

	// Acquire and reuse indicateBuffer. Release it after usage.
	iBuf := <-s.chIndBuf
	defer func() { s.chIndBuf <- iBuf }()

	rsp := HandleValueIndication(iBuf)
	rsp.SetAttributeOpcode()
	rsp.SetAttributeHandle(h)
	buf := bytes.NewBuffer(rsp.AttributeValue())
	buf.Reset()
	if len(data) > buf.Cap() {
		data = data[:buf.Cap()]
	}
	buf.Write(data)
	n, err := s.l2c.Write(rsp[:3+buf.Len()])
	if err != nil {
		return n, err
	}
	select {
	case ok := <-s.chConfirm:
		if !ok {
			return 0, io.ErrClosedPipe
		}
		return n, nil
	case <-time.After(time.Second * 30):
		return 0, ErrSeqProtoTimeout
	}
}

// Loop accepts incoming ATT request, and respond response.
func (s *Server) Loop() {
	for {
		n, err := s.l2c.Read(s.rxBuf)
		if n == 0 || err != nil {
			s.close()
			break
		}
		if rsp := s.handleRequest(s.rxBuf[:n]); rsp != nil {
			if len(rsp) != 0 {
				s.l2c.Write(rsp)
			}
		}
	}
}

func (s *Server) close() error {
	s.chConfirm <- false
	return s.l2c.Close()
}

func (s *Server) handleRequest(b []byte) []byte {
	var resp []byte
	// log.Printf("att req: % X", b)
	switch reqType := b[0]; reqType {
	case ExchangeMTURequestCode:
		resp = s.handleExchangeMTURequest(b)
	case FindInformationRequestCode:
		resp = s.handleFindInformationRequest(b)
	case FindByTypeValueRequestCode:
		resp = s.handleFindByTypeValueRequest(b)
	case ReadByTypeRequestCode:
		resp = s.handleReadByTypeRequest(b)
	case ReadRequestCode:
		resp = s.handleReadRequest(b)
	case ReadBlobRequestCode:
		resp = s.handleReadBlobRequest(b)
	case ReadByGroupTypeRequestCode:
		resp = s.handleReadByGroupRequest(b)
	case WriteRequestCode:
		resp = s.handleWriteRequest(b)
	case WriteCommandCode:
		s.handleWriteCommand(b)
	case ReadMultipleRequestCode,
		PrepareWriteRequestCode,
		ExecuteWriteRequestCode,
		SignedWriteCommandCode:
		fallthrough
	case HandleValueConfirmationCode:
		select {
		case s.chConfirm <- true:
		default:
			log.Printf("recieved a spurious confirmation")
		}
	default:
		resp = NewErrorResponse(reqType, 0x0000, bt.ErrReqNotSupp)
	}
	// log.Printf("att: rsp: % X", resp)
	return resp
}

// handle MTU Exchange request. [Vol 3, Part F, 3.4.2]
func (s *Server) handleExchangeMTURequest(r ExchangeMTURequest) []byte {
	// Validate the request.
	switch {
	case len(r) != 3:
		fallthrough
	case r.ClientRxMTU() < 23:
		return NewErrorResponse(r.AttributeOpcode(), 0x0000, bt.ErrInvalidPDU)
	}

	txMTU := int(r.ClientRxMTU())
	s.l2c.SetTxMTU(txMTU)
	s.l2c.SetRxMTU(len(s.rxBuf))

	if txMTU != len(s.txBuf) {
		// Apply the txMTU afer this response has been sent and before
		// any other attribute protocol PDU is sent.
		defer func() {
			s.txBuf = make([]byte, txMTU, txMTU)
			<-s.chNotBuf
			s.chNotBuf <- make([]byte, txMTU, txMTU)
			<-s.chIndBuf
			s.chIndBuf <- make([]byte, txMTU, txMTU)
		}()
	}

	rsp := ExchangeMTUResponse(s.txBuf)
	rsp.SetAttributeOpcode()
	rsp.SetServerRxMTU(uint16(len(s.rxBuf)))
	return rsp[:3]
}

// handle Find Information request. [Vol 3, Part F, 3.4.3.1 & 3.4.3.2]
func (s *Server) handleFindInformationRequest(r FindInformationRequest) []byte {
	// Validate the request.
	switch {
	case len(r) != 5:
		return NewErrorResponse(r.AttributeOpcode(), 0x0000, bt.ErrInvalidPDU)
	case r.StartingHandle() == 0 || r.StartingHandle() > r.EndingHandle():
		return NewErrorResponse(r.AttributeOpcode(), r.StartingHandle(), bt.ErrInvalidHandle)
	}

	rsp := FindInformationResponse(s.txBuf)
	rsp.SetAttributeOpcode()
	rsp.SetFormat(0x00)
	buf := bytes.NewBuffer(rsp.InformationData())
	buf.Reset()

	// Each response shall contain Types of the same format.
	for _, a := range s.attrs.subrange(r.StartingHandle(), r.EndingHandle()) {
		if rsp.Format() == 0 {
			rsp.SetFormat(0x01)
			if a.Type().Len() == 16 {
				rsp.SetFormat(0x02)
			}<commit_msg>att: rework req/rsp handling loop<commit_after>package att

import (
	"bytes"
	"encoding/binary"
	"io"
	"log"
	"time"

	"github.com/currantlabs/bt"
	"github.com/currantlabs/bt/uuid"
)

// Server implementas an ATT (Attribute Protocol) server.
type Server struct {
	l2c   bt.Conn
	attrs *Range

	// Refer to [Vol 3, Part F, 3.3.2 & 3.3.3] for the requirement of
	// sequential request-response protocol, and transactions.
	rxMTU     int
	txBuf     []byte
	chNotBuf  chan []byte
	chIndBuf  chan []byte
	chConfirm chan bool
}

// NewServer returns an ATT (Attribute Protocol) server.
func NewServer(a *Range, l2c bt.Conn, rxMTU int) *Server {
	// Although the rxBuf is initialized with the capacity of rxMTU, it is
	// not discovered, and only the default ATT_MTU (23 bytes) of it shall
	// be used until remote central request ExchangeMTU.
	s := &Server{
		l2c:   l2c,
		attrs: a,

		rxMTU:     rxMTU,
		txBuf:     make([]byte, 23, 23),
		chNotBuf:  make(chan []byte, 1),
		chIndBuf:  make(chan []byte, 1),
		chConfirm: make(chan bool, 1),
	}
	s.chNotBuf <- make([]byte, 23, 23)
	s.chIndBuf <- make([]byte, 23, 23)
	return s
}

// Notify sends notification to remote central.
func (s *Server) Notify(ind bool, h uint16, data []byte) (int, error) {
	if ind {
		return s.indicate(h, data)
	}
	return s.notify(h, data)
}

// notify sends notification to remote central.
func (s *Server) notify(h uint16, data []byte) (int, error) {
	// log.Printf("att: notifying 0x%04X, %s", h, string(data))

	// Acquire and reuse notifyBuffer. Release it after usage.
	nBuf := <-s.chNotBuf
	defer func() { s.chNotBuf <- nBuf }()

	rsp := HandleValueNotification(nBuf)
	rsp.SetAttributeOpcode()
	rsp.SetAttributeHandle(h)
	buf := bytes.NewBuffer(rsp.AttributeValue())
	buf.Reset()
	if len(data) > buf.Cap() {
		data = data[:buf.Cap()]
	}
	buf.Write(data)
	return s.l2c.Write(rsp[:3+buf.Len()])
}

// indicate sends indication to remote central.
func (s *Server) indicate(h uint16, data []byte) (int, error) {
	// log.Printf("att: indicating 0x%04X, %s", h, string(data))

	// Acquire and reuse indicateBuffer. Release it after usage.
	iBuf := <-s.chIndBuf
	defer func() { s.chIndBuf <- iBuf }()

	rsp := HandleValueIndication(iBuf)
	rsp.SetAttributeOpcode()
	rsp.SetAttributeHandle(h)
	buf := bytes.NewBuffer(rsp.AttributeValue())
	buf.Reset()
	if len(data) > buf.Cap() {
		data = data[:buf.Cap()]
	}
	buf.Write(data)
	n, err := s.l2c.Write(rsp[:3+buf.Len()])
	if err != nil {
		return n, err
	}
	select {
	case ok := <-s.chConfirm:
		if !ok {
			return 0, io.ErrClosedPipe
		}
		return n, nil
	case <-time.After(time.Second * 30):
		return 0, ErrSeqProtoTimeout
	}
}

// Loop accepts incoming ATT request, and respond response.
func (s *Server) Loop() {
	type sbuf struct {
		buf []byte
		len int
	}
	pool := make(chan *sbuf, 2)
	pool <- &sbuf{buf: make([]byte, s.rxMTU)}
	pool <- &sbuf{buf: make([]byte, s.rxMTU)}

	seq := make(chan *sbuf)
	go func() {
		b := <-pool
		for {
			n, err := s.l2c.Read(b.buf)
			if n == 0 || err != nil {
				close(seq)
				s.close()
				return
			}
			if b.buf[0] == HandleValueConfirmationCode {
				select {
				case s.chConfirm <- true:
				default:
					log.Printf("svr: recieved a spurious confirmation")
				}
				continue
			}
			b.len = n
			seq <- b   // Send the current request for handling
			b = <-pool // Swap the buffer for next incoming request.
		}
	}()
	for req := range seq {
		if rsp := s.handleRequest(req.buf[:req.len]); rsp != nil {
			if len(rsp) != 0 {
				s.l2c.Write(rsp)
			}
		}
		pool <- req
	}
}

func (s *Server) close() error {
	s.chConfirm <- false
	return s.l2c.Close()
}

func (s *Server) handleRequest(b []byte) []byte {
	var resp []byte
	// log.Printf("att req: % X", b)
	switch reqType := b[0]; reqType {
	case ExchangeMTURequestCode:
		resp = s.handleExchangeMTURequest(b)
	case FindInformationRequestCode:
		resp = s.handleFindInformationRequest(b)
	case FindByTypeValueRequestCode:
		resp = s.handleFindByTypeValueRequest(b)
	case ReadByTypeRequestCode:
		resp = s.handleReadByTypeRequest(b)
	case ReadRequestCode:
		resp = s.handleReadRequest(b)
	case ReadBlobRequestCode:
		resp = s.handleReadBlobRequest(b)
	case ReadByGroupTypeRequestCode:
		resp = s.handleReadByGroupRequest(b)
	case WriteRequestCode:
		resp = s.handleWriteRequest(b)
	case WriteCommandCode:
		s.handleWriteCommand(b)
	case ReadMultipleRequestCode,
		PrepareWriteRequestCode,
		ExecuteWriteRequestCode,
		SignedWriteCommandCode:
		fallthrough
	default:
		resp = NewErrorResponse(reqType, 0x0000, bt.ErrReqNotSupp)
	}
	// log.Printf("att: rsp: % X", resp)
	return resp
}

// handle MTU Exchange request. [Vol 3, Part F, 3.4.2]
func (s *Server) handleExchangeMTURequest(r ExchangeMTURequest) []byte {
	// Validate the request.
	switch {
	case len(r) != 3:
		fallthrough
	case r.ClientRxMTU() < 23:
		return NewErrorResponse(r.AttributeOpcode(), 0x0000, bt.ErrInvalidPDU)
	}

	txMTU := int(r.ClientRxMTU())
	s.l2c.SetTxMTU(txMTU)
	s.l2c.SetRxMTU(s.rxMTU)

	if txMTU != len(s.txBuf) {
		// Apply the txMTU afer this response has been sent and before
		// any other attribute protocol PDU is sent.
		defer func() {
			s.txBuf = make([]byte, txMTU, txMTU)
			<-s.chNotBuf
			s.chNotBuf <- make([]byte, txMTU, txMTU)
			<-s.chIndBuf
			s.chIndBuf <- make([]byte, txMTU, txMTU)
		}()
	}

	rsp := ExchangeMTUResponse(s.txBuf)
	rsp.SetAttributeOpcode()
	rsp.SetServerRxMTU(uint16(s.rxMTU))
	return rsp[:3]
}

// handle Find Information request. [Vol 3, Part F, 3.4.3.1 & 3.4.3.2]
func (s *Server) handleFindInformationRequest(r FindInformationRequest) []byte {
	// Validate the request.
	switch {
	case len(r) != 5:
		return NewErrorResponse(r.AttributeOpcode(), 0x0000, bt.ErrInvalidPDU)
	case r.StartingHandle() == 0 || r.StartingHandle() > r.EndingHandle():
		return NewErrorResponse(r.AttributeOpcode(), r.StartingHandle(), bt.ErrInvalidHandle)
	}

	rsp := FindInformationResponse(s.txBuf)
	rsp.SetAttributeOpcode()
	rsp.SetFormat(0x00)
	buf := bytes.NewBuffer(rsp.InformationData())
	buf.Reset()

	// Each response shall contain Types of the same format.
	for _, a := range s.attrs.subrange(r.StartingHandle(), r.EndingHandle()) {
		if rsp.Format() == 0 {
			rsp.SetFormat(0x01)
			if a.Type().Len() == 16 {
				rsp.SetFormat(0x02)
			}
{'fraction_non_alphanumeric': 0.14441958737260752, 'fraction_numerical': 0.017399950285856326, 'mean_word_length': 3.729623824451411, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6458353', 'n_tokens_mistral': 4882, 'n_tokens_neox': 4536, 'n_words': 1155}