Method RoutineSequentialReceiver

polyamide/device/receive.go:428–510

(maxBatchSize int)

}

func (peer *Peer) RoutineSequentialReceiver(maxBatchSize int) {
	device := peer.device
	defer func() {
		device.Log.Verbosef("%v - Routine: sequential receiver - stopped", peer)
		peer.stopping.Done()
	}()
	device.Log.Verbosef("%v - Routine: sequential receiver - started", peer)

	var (
		tcBufs = make([]*TCElement, 0, maxBatchSize)
		tcs    = NewTCState()
	)

	for elemsContainer := range peer.queue.inbound.c {
		if elemsContainer == nil {
			return
		}
		elemsContainer.Lock()
		validTailPacket := -1
		dataPacketReceived := false
		rxBytesLen := uint64(0)
		rxPkts := 0

		for i, elem := range elemsContainer.elems {
			if elem.packet == nil {
				// decryption failed
				continue
			}

			if !elem.keypair.replayFilter.ValidateCounter(elem.counter, RejectAfterMessages) {
				continue
			}

			validTailPacket = i
			if peer.ReceivedWithKeypair(elem.keypair) {
				peer.SetEndpointFromPacket(elem.endpoint)
				peer.timersHandshakeComplete()
				peer.SendStagedPackets()
			}
			rxBytesLen += uint64(len(elem.packet) + MinMessageSize)
			rxPkts++

			if len(elem.packet) == 0 {
				device.Log.Verbosef("%v - Receiving keepalive packet", peer)
				continue
			}
			dataPacketReceived = true

			tce := device.GetTCElement()
			tce.Packet = elem.packet
			tce.Buffer = elem.buffer
			elem.buffer = nil
			elem.packet = nil
			tce.FromEp = elem.endpoint
			tce.FromPeer = peer

			tcBufs = append(tcBufs, tce)
		}