Method BatchAndSendMessages

conn/node.go:338–396

BatchAndSendMessages sends messages in batches.

()

// BatchAndSendMessages sends messages in batches.
func (n *Node) BatchAndSendMessages() {
	batches := make(map[uint64]*bytes.Buffer)
	streams := make(map[uint64]*stream)

	for {
		totalSize := 0
		sm := <-n.messages
	slurp_loop:
		for {
			var buf *bytes.Buffer
			if b, ok := batches[sm.to]; !ok {
				buf = new(bytes.Buffer)
				batches[sm.to] = buf
			} else {
				buf = b
			}
			totalSize += 4 + len(sm.data)
			x.Check(binary.Write(buf, binary.LittleEndian, uint32(len(sm.data))))
			x.Check2(buf.Write(sm.data))

			if totalSize > messageBatchSoftLimit {
				// We limit the batch size, but we aren't pushing back on
				// n.messages, because the loop below spawns a goroutine
				// to do its dirty work.  This is good because right now
				// (*node).send fails(!) if the channel is full.
				break
			}

			select {
			case sm = <-n.messages:
			default:
				break slurp_loop
			}
		}

		for to, buf := range batches {
			if buf.Len() == 0 {
				continue
			}
			s, ok := streams[to]
			if !ok || atomic.LoadInt32(&s.alive) <= 0 {
				s = &stream{
					msgCh: make(chan []byte, 100),
					alive: 1,
				}
				go n.streamMessages(to, s)
				streams[to] = s
			}
			data := make([]byte, buf.Len())
			copy(data, buf.Bytes())
			buf.Reset()

			select {
			case s.msgCh <- data:
			default:
			}
		}
	}