Method receive

io.go:172–292

receive 用于接收发布或者订阅

(streamPath string, specific common.IIO)

// receive 用于接收发布或者订阅
func (io *IO) receive(streamPath string, specific common.IIO) error {
	streamPath = strings.Trim(streamPath, "/")
	u, err := url.Parse(streamPath)
	if err != nil {
		if EngineConfig.LogLang == "zh" {
			log.Error("接收流路径(流唯一标识)格式错误,必须形如 live/test ", zap.String("流路径", streamPath), zap.Error(err))
		} else {
			log.Error("receive streamPath wrong format", zap.String("streamPath", streamPath), zap.Error(err))
		}
		return err
	}
	io.Args = u.Query()
	wt := time.Second * 5
	var iSub ISubscriber
	var iPub IPublisher
	var isSubscribe bool
	if iSub, isSubscribe = specific.(ISubscriber); isSubscribe {
		wt = iSub.GetSubscriber().Config.WaitTimeout
	} else {
		iPub = specific.(IPublisher)
	}
	s, create := findOrCreateStream(u.Path, wt)
	if s == nil {
		return ErrBadStreamName
	}
	if io.Stream == nil { //初次
		if io.Type == "" {
			io.Type = reflect.TypeOf(specific).Elem().Name()
		}
		logFeilds := []zapcore.Field{zap.String("type", io.Type)}
		if io.ID != "" {
			logFeilds = append(logFeilds, zap.String("ID", io.ID))
		}
		if io.Logger == nil {
			io.Logger = s.With(logFeilds...)
		} else {
			io.Logger = io.Logger.With(logFeilds...)
		}
	}
	io.Stream = s
	io.Spesific = specific
	io.StartTime = time.Now()
	if io.Context == nil {
		io.Debug("create context")
		io.SetParentCtx(Engine.Context)
	} else if io.IsClosed() {
		io.Debug("recreate context")
		io.SetParentCtx(Engine.Context)
	} else {
		io.Debug("warp context")
		io.SetParentCtx(io.Context)
	}
	defer func() {
		if err == nil {
			specific.OnEvent(specific)
		}
	}()
	if !isSubscribe {