Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use checkClients method of main Package

Best Syzkaller code snippet using main.checkClients

streamCore.go

Source:streamCore.go

...59	if url, err := url.Parse(opt.URL); err == nil && strings.ToLower(url.Scheme) == "rtmp" {60		return StreamServerRunStreamRTMP(streamID, channelID, opt)61	}62	keyTest := time.NewTimer(20 * time.Second)63	checkClients := time.NewTimer(20 * time.Second)64	var start bool65	var fps int66	var preKeyTS = time.Duration(0)67	var Seq []*av.Packet68	RTSPClient, err := rtspv2.Dial(rtspv2.RTSPClientOptions{URL: opt.URL, InsecureSkipVerify: opt.InsecureSkipVerify, DisableAudio: !opt.Audio, DialTimeout: 3 * time.Second, ReadWriteTimeout: 5 * time.Second, Debug: opt.Debug, OutgoingProxy: true})69	if err != nil {70		return 0, err71	}72	Storage.StreamChannelStatus(streamID, channelID, ONLINE)73	defer func() {74		RTSPClient.Close()75		Storage.StreamChannelStatus(streamID, channelID, OFFLINE)76		Storage.StreamHLSFlush(streamID, channelID)77	}()78	var WaitCodec bool79	/*80		Example wait codec81	*/82	if RTSPClient.WaitCodec {83		WaitCodec = true84	} else {85		if len(RTSPClient.CodecData) > 0 {86			Storage.StreamChannelCodecsUpdate(streamID, channelID, RTSPClient.CodecData, RTSPClient.SDPRaw)87		}88	}89	log.WithFields(logrus.Fields{90		"module":  "core",91		"stream":  streamID,92		"channel": channelID,93		"func":    "StreamServerRunStream",94		"call":    "Start",95	}).Infoln("Success connection RTSP")96	var ProbeCount int97	var ProbeFrame int98	var ProbePTS time.Duration99	Storage.NewHLSMuxer(streamID, channelID)100	defer Storage.HLSMuxerClose(streamID, channelID)101	for {102		select {103		//Check stream have clients104		case <-checkClients.C:105			if opt.OnDemand && !Storage.ClientHas(streamID, channelID) {106				return 1, ErrorStreamNoClients107			}108			checkClients.Reset(20 * time.Second)109		//Check stream send key110		case <-keyTest.C:111			return 0, ErrorStreamNoVideo112		//Read core signals113		case signals := <-opt.signals:114			switch signals {115			case SignalStreamStop:116				return 2, ErrorStreamStopCoreSignal117			case SignalStreamRestart:118				return 0, ErrorStreamRestart119			case SignalStreamClient:120				return 1, ErrorStreamNoClients121			}122		//Read rtsp signals123		case signals := <-RTSPClient.Signals:124			switch signals {125			case rtspv2.SignalCodecUpdate:126				Storage.StreamChannelCodecsUpdate(streamID, channelID, RTSPClient.CodecData, RTSPClient.SDPRaw)127				WaitCodec = false128			case rtspv2.SignalStreamRTPStop:129				return 0, ErrorStreamStopRTSPSignal130			}131		case packetRTP := <-RTSPClient.OutgoingProxyQueue:132			Storage.StreamChannelCastProxy(streamID, channelID, packetRTP)133		case packetAV := <-RTSPClient.OutgoingPacketQueue:134			if WaitCodec {135				continue136			}137			if packetAV.IsKeyFrame {138				keyTest.Reset(20 * time.Second)139				if preKeyTS > 0 {140					Storage.StreamHLSAdd(streamID, channelID, Seq, packetAV.Time-preKeyTS)141					Seq = []*av.Packet{}142				}143				preKeyTS = packetAV.Time144			}145			Seq = append(Seq, packetAV)146			Storage.StreamChannelCast(streamID, channelID, packetAV)147			/*148			   HLS LL Test149			*/150			if packetAV.IsKeyFrame && !start {151				start = true152			}153			/*154				FPS mode probe155			*/156			if start {157				ProbePTS += packetAV.Duration158				ProbeFrame++159				if packetAV.IsKeyFrame && ProbePTS.Seconds() >= 1 {160					ProbeCount++161					if ProbeCount == 2 {162						fps = int(math.Round(float64(ProbeFrame) / ProbePTS.Seconds()))163					}164					ProbeFrame = 0165					ProbePTS = 0166				}167			}168			if start && fps != 0 {169				//TODO fix it170				packetAV.Duration = time.Duration((float32(1000)/float32(fps))*1000*1000) * time.Nanosecond171				Storage.HlsMuxerSetFPS(streamID, channelID, fps)172				Storage.HlsMuxerWritePacket(streamID, channelID, packetAV)173			}174		}175	}176}177func StreamServerRunStreamRTMP(streamID string, channelID string, opt *ChannelST) (int, error) {178	keyTest := time.NewTimer(20 * time.Second)179	checkClients := time.NewTimer(20 * time.Second)180	OutgoingPacketQueue := make(chan *av.Packet, 1000)181	Signals := make(chan int, 100)182	var start bool183	var fps int184	var preKeyTS = time.Duration(0)185	var Seq []*av.Packet186	conn, err := rtmp.DialTimeout(opt.URL, 3*time.Second)187	if err != nil {188		return 0, err189	}190	Storage.StreamChannelStatus(streamID, channelID, ONLINE)191	defer func() {192		conn.Close()193		Storage.StreamChannelStatus(streamID, channelID, OFFLINE)194		Storage.StreamHLSFlush(streamID, channelID)195	}()196	var WaitCodec bool197	codecs, err := conn.Streams()198	if err != nil {199		return 0, err200	}201	preDur := make([]time.Duration, len(codecs))202	Storage.StreamChannelCodecsUpdate(streamID, channelID, codecs, []byte{})203	log.WithFields(logrus.Fields{204		"module":  "core",205		"stream":  streamID,206		"channel": channelID,207		"func":    "StreamServerRunStream",208		"call":    "Start",209	}).Infoln("Success connection RTSP")210	var ProbeCount int211	var ProbeFrame int212	var ProbePTS time.Duration213	Storage.NewHLSMuxer(streamID, channelID)214	defer Storage.HLSMuxerClose(streamID, channelID)215	go func() {216		for {217			ptk, err := conn.ReadPacket()218			if err != nil {219				break220			}221			OutgoingPacketQueue <- &ptk222		}223		Signals <- 1224	}()225	for {226		select {227		//Check stream have clients228		case <-checkClients.C:229			if opt.OnDemand && !Storage.ClientHas(streamID, channelID) {230				return 1, ErrorStreamNoClients231			}232			checkClients.Reset(20 * time.Second)233		//Check stream send key234		case <-keyTest.C:235			return 0, ErrorStreamNoVideo236		//Read core signals237		case signals := <-opt.signals:238			switch signals {239			case SignalStreamStop:240				return 2, ErrorStreamStopCoreSignal241			case SignalStreamRestart:242				return 0, ErrorStreamRestart243			case SignalStreamClient:244				return 1, ErrorStreamNoClients245			}246		//Read rtsp signals...

airsignals.go

Source:airsignals.go

...51		},52		MaxAge: 12 * time.Hour,53	}))54	router.GET("/ws/:chatID/:hostID", socket)55	router.GET("/getConnectedClients/:chatID", checkClients)56	// Run signaling server57	if localhostflag {58		router.Run("localhost:8080")59	}60	router.Run("0.0.0.0:8080")61}62func checkClients(c *gin.Context) {63	chatID := c.Param("chatID")64	_, ok := threadSafeRooms.chatRooms[chatID]65	if !ok {66		c.JSON(401, gin.H{67			"type": "message",68			"body": "Chat Room does not exist",69		})70	} else {71		c.JSON(200, gin.H{72			"type":       "message",73			"numClients": threadSafeRooms.chatRooms[chatID].GetNumClients(),74			"body":       "",75		})76	}...

pool.go

Source:pool.go

...22	}23	return &pool24}25// Starts the pool. Pool listens now to incoming tcp connections and adds new clients to pClients.26// Also starts a goroutine that pings every client every `checkClientsDuration` seconds. If they do not respond, they will be removed.27func (pool *Pool) Start() error {28	pool.Running = true29	go pool.checkClients()30	listener, err := net.Listen("tcp", fmt.Sprintf(":%s", pool.Port))31	pool.listener = listener32	// check if initiating the listener failed33	if err != nil {34		return err35	}36	// main loop to check of incoming connections37	for {38		// accepting an incoming connection39		conn, err := listener.Accept()40		// if accepting the connection failes, just ignore it41		if err != nil {42			continue43		}44		// read stream until \n45		netData, err := bufio.NewReader(conn).ReadString('\n')46		// if reading the incoming data failes, just ignore the connection47		if err != nil {48			continue49		} else {50			// a client wants to login51			// append client to clients52			if strings.HasPrefix(netData, "marta login|") {53				client := NewClient(conn)54				clientInfo, err := NewClientInfo(strings.TrimPrefix(netData, "marta login|"))55				if err != nil {56					logger.Println(fmt.Sprintf("Invalid login message: '%s'", netData))57					continue58				}59				client.Info = *clientInfo60				client.Send("marta logged in\n")61				clients.AddClient(*client)62				logger.Println("Client connected: " + client.String())63			} else {64				logger.Println("Received in pool (should not happen): '" + netData + "'")65			}66		}67	}68}69// Runs forever and pings clients every `checkClientsDuration` seconds.70func (pool *Pool) checkClients() {71	dur, err := strconv.Atoi(os.Getenv("CHECK_CLIENTS_DURATION"))72	if err != nil {73		logger.Panicln(err)74	}75	for {76		clients.Ping()77		time.Sleep(time.Duration(dur) * time.Second)78	}79}80// Stop the pool.81func (pool *Pool) Stop() error {82	err := pool.listener.Close()83	pool.Running = false84	return err...

checkClients

Using AI Code Generation

1import (2func main() {3	m := &Main{}4	m.checkClients()5}6type Main struct {7}8type Client struct {9}10func (m *Main) checkClients() {11	c := &Client{}12	c.CreatedAt = time.Now()13	fmt.Println(c)14}15&{1 John Doe 2018-08-25 14:29:20.935 +0000 UTC m=+0.000000001}

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.