// Recv receives a message.  For AF_SP_RAW messages the header data will
// be included at he start of the returned byte slice (otherwise it will
// be stripped).  At this time no flags are supported.
func (s *Socket) Recv(flags int) ([]byte, error) {
	var b []byte
	var m *mangos.Message
	var err error

	if flags != 0 {
		return nil, errNoFlag
	}

	// Legacy nanomsg uses the opposite semantic for negative and
	// zero values than mangos.  A bit unfortunate.
	switch {
	case s.rto > 0:
		s.sock.SetOption(mangos.OptionRecvDeadline, s.rto)
	case s.rto == 0:
		s.sock.SetOption(mangos.OptionRecvDeadline, -1)
	case s.rto < 0:
		s.sock.SetOption(mangos.OptionRecvDeadline, 0)
	}

	if m, err = s.sock.RecvMsg(); err != nil {
		return nil, err
	}

	if s.dom == AF_SP_RAW {
		b = make([]byte, 0, len(m.Body)+len(m.Header))
		b = append(b, m.Header...)
		b = append(b, m.Body...)
	} else {
		b = make([]byte, 0, len(m.Body))
		b = append(b, m.Body...)
	}
	m.Free()
	return b, nil
}

func (r *req) SendHook(m *mangos.Message) bool {

	if r.raw {
		// Raw mode has no automatic retry, and must include the
		// request id in the header coming down.
		return true
	}
	r.Lock()
	defer r.Unlock()

	// We need to generate a new request id, and append it to the header.
	r.reqid = r.nextID()
	v := r.reqid
	m.Header = append(m.Header,
		byte(v>>24), byte(v>>16), byte(v>>8), byte(v))

	r.reqmsg = m.Dup()

	// Schedule a retry, in case we don't get a reply.
	if r.retry > 0 {
		r.waker.Reset(r.retry)
	} else {
		r.waker.Stop()
	}

	r.sock.SetRecvError(nil)

	return true
}

func (w *wsPipe) Send(m *mangos.Message) error {

	var buf []byte

	if len(m.Header) > 0 {
		buf = make([]byte, 0, len(m.Header)+len(m.Body))
		buf = append(buf, m.Header...)
		buf = append(buf, m.Body...)
	} else {
		buf = m.Body
	}
	if err := w.ws.WriteMessage(w.dtype, buf); err != nil {
		return err
	}
	m.Free()
	return nil
}

func (pt *pubTest) SendHook(m *mangos.Message) bool {
	if pt.pubidx >= len(publish) {
		pt.Errorf("Nothing left to send! (%d/%d)", pt.pubidx, len(publish))
		return false
	}
	m.Body = append(m.Body, []byte(publish[pt.pubidx])...)
	pt.Debugf("Sending %d, %s", pt.pubidx, string(m.Body))
	pt.pubidx++
	return pt.T.SendHook(m)
}

func (x *star) SendHook(m *mangos.Message) bool {

	if x.raw {
		// TTL header must be present.
		return true
	}
	// new message has a zero hop count
	m.Header = append(m.Header, 0, 0, 0, 0)
	return true
}

// We save the backtrace from this message.  This means that if the app calls
// Recv before calling Send, the saved backtrace will be lost.  This is how
// the application discards / cancels a request to which it declines to reply.
// This is only done in cooked mode.
func (r *rep) RecvHook(m *mangos.Message) bool {
	if r.raw {
		return true
	}
	r.sock.SetSendError(nil)
	r.backtraceL.Lock()
	r.backtrace = append(r.backtracebuf[0:0], m.Header...)
	r.backtraceL.Unlock()
	m.Header = nil
	return true
}

func (x *bus) broadcast(m *mangos.Message, sender uint32) {

	x.Lock()
	for id, pe := range x.peers {
		if sender == id {
			continue
		}
		m = m.Dup()

		select {
		case pe.q <- m:
		default:
			// No room on outbound queue, drop it.
			// Note that if we are passing on a linger/shutdown
			// notification and we can't deliver due to queue
			// full, it means we will wind up waiting the full
			// linger time in the lower sender.  Its correct, if
			// suboptimal, behavior.
			m.Free()
		}
	}
	x.Unlock()
}

func serverWorker(sock mangos.Socket, id int) {
	var err error

	delay := rand.Intn(int(time.Second))

	for {
		var m *mangos.Message

		if m, err = sock.RecvMsg(); err != nil {
			return
		}

		m.Body = make([]byte, 4)

		time.Sleep(time.Duration(delay))

		binary.BigEndian.PutUint32(m.Body[0:], uint32(id))

		if err = sock.SendMsg(m); err != nil {
			return
		}
	}
}

func (x *star) broadcast(m *mangos.Message, sender *starEp) {

	x.Lock()
	if sender == nil || !x.raw {
		for _, pe := range x.eps {
			if sender == pe {
				continue
			}
			m = m.Dup()
			select {
			case pe.q <- m:
			default:
				// No room on outbound queue, drop it.
				if m != nil {
					m.Free()
				}
			}
		}
	}
	x.Unlock()

	// Grab a local copy and send it up if we aren't originator
	if sender != nil {
		select {
		case x.sock.RecvChannel() <- m:
		case <-x.sock.CloseChannel():
			m.Free()
			return
		default:
			// No room, so we just drop it.
			m.Free()
		}
	} else {
		// Not sending it up, so we need to release it.
		m.Free()
	}
}

func (x *resp) SendHook(m *mangos.Message) bool {
	if x.raw {
		// Raw mode senders expected to have prepared header already.
		return true
	}
	x.sock.SetSendError(mangos.ErrProtoState)
	x.Lock()
	m.Header = append(m.Header[0:0], x.backtrace...)
	x.backtrace = nil
	x.Unlock()
	if len(m.Header) == 0 {
		return false
	}
	return true
}

func (x *surveyor) sender() {
	defer x.w.Done()
	sq := x.sock.SendChannel()
	cq := x.sock.CloseChannel()
	for {
		var m *mangos.Message
		select {
		case m = <-sq:
		case <-cq:
			return
		}

		x.Lock()
		for _, pe := range x.peers {
			m := m.Dup()
			select {
			case pe.q <- m:
			default:
				m.Free()
			}
		}
		x.Unlock()
	}
}

func (x *surveyor) RecvHook(m *mangos.Message) bool {
	if x.raw {
		return true
	}

	x.Lock()
	defer x.Unlock()

	if len(m.Header) < 4 {
		return false
	}
	if binary.BigEndian.Uint32(m.Header) != x.surveyID {
		return false
	}
	m.Header = m.Header[4:]
	return true
}

func (bt *busTest) SendHook(m *mangos.Message) bool {
	bt.Lock()
	defer bt.Unlock()
	v := uint32(bt.GetID())
	w := bt.send
	bt.send++
	m.Body = m.Body[0:8]

	binary.BigEndian.PutUint32(m.Body, v)
	binary.BigEndian.PutUint32(m.Body[4:], w)

	// Inject a sleep to avoid overwhelming the bus and dropping messages.
	//d := time.Duration(rand.Uint32() % 10000)
	//time.Sleep(d * time.Microsecond)

	return bt.T.SendHook(m)
}

func (r *rep) SendHook(m *mangos.Message) bool {
	// Store our saved backtrace.  Note that if none was previously stored,
	// there is no one to reply to, and we drop the message.  We only
	// do this in cooked mode.
	if r.raw {
		return true
	}
	r.sock.SetSendError(mangos.ErrProtoState)
	r.backtraceL.Lock()
	m.Header = append(m.Header[0:0], r.backtrace...)
	r.backtrace = nil
	r.backtraceL.Unlock()
	if m.Header == nil {
		return false
	}
	return true
}

func (r *rep) sender() {
	defer r.w.Done()
	sq := r.sock.SendChannel()
	cq := r.sock.CloseChannel()

	for {
		var m *mangos.Message

		select {
		case m = <-sq:
		case <-cq:
			return
		}

		// Lop off the 32-bit peer/pipe ID.  If absent, drop.
		if len(m.Header) < 4 {
			m.Free()
			continue
		}
		id := binary.BigEndian.Uint32(m.Header)
		m.Header = m.Header[4:]
		r.Lock()
		pe := r.eps[id]
		r.Unlock()
		if pe == nil {
			m.Free()
			continue
		}

		select {
		case pe.q <- m:
		default:
			// If our queue is full, we have no choice but to
			// throw it on the floor.  This shoudn't happen,
			// since each partner should be running synchronously.
			// Devices are a different situation, and this could
			// lead to lossy behavior there.  Initiators will
			// resend if this happens.  Devices need to have deep
			// enough queues and be fast enough to avoid this.
			m.Free()
		}
	}
}

func (x *resp) sender() {
	// This is pretty easy because we have only one peer at a time.
	// If the peer goes away, we'll just drop the message on the floor.

	defer x.w.Done()
	cq := x.sock.CloseChannel()
	sq := x.sock.SendChannel()
	for {
		var m *mangos.Message
		select {
		case m = <-sq:
		case <-cq:
			return
		}

		// Lop off the 32-bit peer/pipe ID.  If absent, drop.
		if len(m.Header) < 4 {
			m.Free()
			continue
		}

		id := binary.BigEndian.Uint32(m.Header)
		m.Header = m.Header[4:]

		x.Lock()
		peer := x.peers[id]
		x.Unlock()

		if peer == nil {
			m.Free()
			continue
		}

		// Put it on the outbound queue
		select {
		case peer.q <- m:
		default:
			// Backpressure, drop it.
			m.Free()
		}
	}
}

func (x *surveyor) SendHook(m *mangos.Message) bool {

	if x.raw {
		return true
	}

	x.Lock()
	x.surveyID = x.nextID | 0x80000000
	x.nextID++
	x.sock.SetRecvError(nil)
	v := x.surveyID
	m.Header = append(m.Header,
		byte(v>>24), byte(v>>16), byte(v>>8), byte(v))

	if x.duration > 0 {
		x.timer.Reset(x.duration)
	}
	x.Unlock()

	return true
}

func (rt *reqTest) SendHook(m *mangos.Message) bool {
	m.Body = append(m.Body, byte(rt.GetSend()))
	rt.tot = rt.GetSend()
	return rt.T.SendHook(m)
}

func (dt *devTest) SendHook(m *mangos.Message) bool {
	m.Body = append(m.Body, byte(dt.GetSend()))
	return dt.T.SendHook(m)
}

func (pt *pairTest) SendHook(m *mangos.Message) bool {
	m.Body = append(m.Body, byte(pt.GetSend()))
	return pt.T.SendHook(m)
}