Method processRecvData

pkg/streamclient/streambroker.go:179–205

(dataPk wshrpc.CommandStreamData)

}

func (b *Broker) processRecvData(dataPk wshrpc.CommandStreamData) {
	b.lock.Lock()
	reader, ok := b.readers[dataPk.Id]
	if !ok {
		lastSent := b.readerErrorSentTime[dataPk.Id]
		now := time.Now()
		if now.Sub(lastSent) < time.Second {
			b.lock.Unlock()
			return
		}
		b.readerErrorSentTime[dataPk.Id] = now
	}
	b.lock.Unlock()

	if !ok {
		ackPk := wshrpc.CommandStreamAckData{
			Id:     dataPk.Id,
			Seq:    dataPk.Seq,
			Cancel: true,
			Error:  "stream reader not found",
		}
		b.SendAck(ackPk)
		return
	}

	reader.RecvData(dataPk)
}

func (b *Broker) processRecvAck(ackPk wshrpc.CommandStreamAckData) {
	b.lock.Lock()