Method run

stream.go:443–688

流状态处理中枢，包括接收订阅发布指令等

()

// 流状态处理中枢，包括接收订阅发布指令等
func (s *Stream) run() {
	EventBus <- SEcreate{StreamEvent{Event[*Stream]{Target: s, Time: time.Now()}}}
	pulseTicker := time.NewTicker(EngineConfig.PulseInterval)
	defer pulseTicker.Stop()
	var timeOutInfo zap.Field
	var timeStart time.Time
	for pulseSuber := make(map[ISubscriber]struct{}); ; s.checkRunCost(timeStart, timeOutInfo) {
		select {
		case <-pulseTicker.C:
			timeStart = time.Now()
			timeOutInfo = zap.String("type", "pulse")
			for sub := range pulseSuber {
				sub.OnEvent(PulseEvent{CreateEvent(struct{}{})})
			}
		case <-s.timeout.C:
			timeStart = time.Now()
			timeOutInfo = zap.String("state", s.State.String())
			if s.State == STATE_PUBLISHING || s.State == STATE_WAITTRACK {
				for sub := range s.Subscribers.internal {
					if sub.IsClosed() {
						delete(s.Subscribers.internal, sub)
						s.Info("innersuber -1", zap.Int("remains", len(s.Subscribers.internal)))
					}
				}
				for sub := range s.Subscribers.public {
					if sub.IsClosed() {
						s.onSuberClose(sub)
					}
				}
				if !s.NeverTimeout {
					lost := false
					trackCount := 0
					timeout := s.PublishTimeout
					if s.IsPause {
						timeout = s.PauseTimeout
					}
					s.Tracks.Range(func(name string, t common.Track) {
						trackCount++
						switch t.(type) {
						case *track.Video, *track.Audio:
							// track 超过一定时间没有更新数据了
							if lastWriteTime := t.LastWriteTime(); !lastWriteTime.IsZero() && time.Since(lastWriteTime) > timeout {
								s.Warn("track timeout", zap.String("name", name), zap.Time("last writetime", lastWriteTime), zap.Duration("timeout", timeout))
								lost = true
							}
						}
					})
					if !lost {
						if trackCount == 0 {
							s.Warn("no tracks")
							if time.Since(s.StartTime) > timeout {
								lost = true
							}
						} else if s.Publisher != nil && s.Publisher.IsClosed() {
							s.Warn("publish is closed", zap.Error(context.Cause(s.publisher)), zap.String("ptr", fmt.Sprintf("%p", s.publisher.Context)))
							lost = true
						}
					}