Method Pull

plugin.go:268–321

(streamPath string, url string, puller IPuller, save int)

var Pullers sync.Map

func (opt *Plugin) Pull(streamPath string, url string, puller IPuller, save int) (err error) {
	conf, ok := opt.Config.(config.PullConfig)
	if !ok {
		return ErrNoPullConfig
	}
	pullConf := conf.GetPullConfig()
	if save < 2 {
		zurl := zap.String("url", url)
		zpath := zap.String("stream", streamPath)
		opt.Info("pull", zpath, zurl)
		puller.init(streamPath, url, pullConf)
		opt.AssignPubConfig(puller.GetPublisher())
		puller.SetLogger(opt.Logger.With(zpath, zurl))
		go puller.startPull(puller)
	}
	switch save {
	case 1:
		pullConf.PullOnStartLocker.Lock()
		defer pullConf.PullOnStartLocker.Unlock()
		m := map[string]string{streamPath: url}
		for id := range pullConf.PullOnStart {
			m[id] = pullConf.PullOnStart[id]
		}
		m[streamPath] = url
		opt.RawConfig.ParseModifyFile(map[string]any{
			"pull": map[string]any{
				"pullonstart": m,
			},
		})
	case 2:
		pullConf.PullOnSubLocker.Lock()
		defer pullConf.PullOnSubLocker.Unlock()
		m := map[string]string{streamPath: url}
		for id := range pullConf.PullOnSub {
			m[id] = pullConf.PullOnSub[id]
		}
		m[streamPath] = url
		if pullConf.PullOnSub == nil {
			pullConf.PullOnSub = make(map[string]string)
		}
		pullConf.PullOnSub[streamPath] = url
		opt.RawConfig.ParseModifyFile(map[string]any{
			"pull": map[string]any{
				"pullonsub": m,
			},
		})
	}
	if save > 0 {
		if err = opt.Save(); err != nil {
			opt.Error("save faild", zap.Error(err))
		}
	}
	return
}

var ErrNoPushConfig = errors.New("no push config")
var Pushers sync.Map