Method dispatcherLoop

lib/protocol/protocol.go:427–504

()

}

func (c *rawConnection) dispatcherLoop() (err error) {
	defer close(c.dispatcherLoopStopped)
	var msg proto.Message
	state := stateInitial
	for {
		select {
		case <-c.closed:
			return ErrClosed
		default:
		}
		select {
		case msg = <-c.inbox:
		case <-c.closed:
			return ErrClosed
		}

		metricDeviceRecvMessages.WithLabelValues(c.idString).Inc()

		msgContext, err := messageContext(msg)
		if err != nil {
			return fmt.Errorf("protocol error: %w", err)
		}
		l.Debugf("handle %v message", msgContext)

		switch msg := msg.(type) {
		case *bep.ClusterConfig:
			if state == stateInitial {
				state = stateReady
			}
		case *bep.Close:
			return fmt.Errorf("closed by remote: %v", msg.Reason)
		default:
			if state != stateReady {
				return newProtocolError(fmt.Errorf("invalid state %d", state), msgContext)
			}
		}

		switch msg := msg.(type) {
		case *bep.ClusterConfig:
			err = c.model.ClusterConfig(clusterConfigFromWire(msg))

		case *bep.Index:
			idx := indexFromWire(msg)
			if err := checkIndexConsistency(idx.Files); err != nil {
				return newProtocolError(err, msgContext)
			}
			err = c.handleIndex(idx)

		case *bep.IndexUpdate:
			idxUp := indexUpdateFromWire(msg)
			if err := checkIndexConsistency(idxUp.Files); err != nil {
				return newProtocolError(err, msgContext)
			}
			err = c.handleIndexUpdate(idxUp)

		case *bep.Request:
			if err := checkFilename(msg.Name); err != nil {
				return newProtocolError(err, msgContext)