Got response on RTSP session.

        Manage time out handle since response came in a reasonable time.
        Update session parameters with latest response.
        If state is playing schedule keep-alive.
557713
def data_received(self, data):
        
        self.time_out_handle.cancel()
        self.session.update(data.decode())

        if self.session.state == STATE_STARTING:
            self.transport.write(self.method.message.encode())
            self.time_out_handle = self.loop.call_later(
                TIME_OUT_LIMIT, self.time_out)

        elif self.session.state == STATE_PLAYING:
            self.callback(SIGNAL_PLAYING)

            if self.session.session_timeout != 0:
                interval = self.session.session_timeout - 5
                self.loop.call_later(interval, self.keep_alive)

        else:
            self.stop()
497534
def respond(response, headers_or_body = {}, body = nil)
      raise StateError, "not in header state" if @response_state != :headers

      if headers_or_body.is_a? Hash
        headers = headers_or_body
      else
        headers = {}
        body = headers_or_body
      end

      if @keepalive
        headers[CONNECTION] = KEEP_ALIVE
      else
        headers[CONNECTION] = CLOSE
      end

      case response
      when Symbol, Fixnum, Integer
        response = Response.new(response, headers, body)
      when Response
      else raise TypeError, "invalid response: #{response.inspect}"
      end

      if current_request
        current_request.handle_response(response)
      else
        raise RequestError
      end

      # Enable streaming mode
      if response.chunked? and response.body.nil?
        @response_state = :chunked_body
      elsif @keepalive
        reset_request
      else
        @current_request = nil
        @parser.reset
        @request_fsm.transition :closed
      end
    rescue IOError, SystemCallError, RequestError
      # The client disconnected early, or there is no request
      @keepalive = false
      @request_fsm.transition :closed
      @parser.reset
      @current_request = nil
    end
632103
def keep_alive_timeout_callback(self):
        
        time_elapsed = time() - self._last_response_time
        if time_elapsed < self.keep_alive_timeout:
            time_left = self.keep_alive_timeout - time_elapsed
            self._keep_alive_timeout_handler = self.loop.call_later(
                time_left, self.keep_alive_timeout_callback
            )
        else:
            logger.debug("KeepAlive Timeout. Closing connection.")
            self.transport.close()
            self.transport = None
1012010
def ping(self):
        

        # Make sure we're connected
        if not self.conn:
            self.connect()

        # Send the ping and wait for the response
        self.conn.send('PING', time.time())
        cmd, payload = self.conn.recv()
        recv_ts = time.time()

        # Make sure the response was a PONG
        if cmd != 'PONG':
            raise Exception("Invalid response from server")

        # Return the RTT
        return recv_ts - payload[0]
595575
def handle_simple_responses(
      self, timeout_ms=None, info_cb=DEFAULT_MESSAGE_CALLBACK):
    
    return self._accept_responses('OKAY', info_cb, timeout_ms=timeout_ms)
1035170
private synchronized void keepAliveSessions(long lastKeepAliveTime, long sessionTimeout) {
    // Filter the list of sessions by timeout.
    List<RaftSessionState> needKeepAlive = sessions.values()
        .stream()
        .filter(session -> session.getSessionTimeout() == sessionTimeout)
        .collect(Collectors.toList());

    // If no sessions need keep-alives to be sent, skip and reschedule the keep-alive.
    if (needKeepAlive.isEmpty()) {
      return;
    }

    // Allocate session IDs, command response sequence numbers, and event index arrays.
    long[] sessionIds = new long[needKeepAlive.size()];
    long[] commandResponses = new long[needKeepAlive.size()];
    long[] eventIndexes = new long[needKeepAlive.size()];

    // For each session that needs to be kept alive, populate batch request arrays.
    int i = 0;
    for (RaftSessionState sessionState : needKeepAlive) {
      sessionIds[i] = sessionState.getSessionId().id();
      commandResponses[i] = sessionState.getCommandResponse();
      eventIndexes[i] = sessionState.getEventIndex();
      i++;
    }

    log.trace("Keeping {} sessions alive", sessionIds.length);

    KeepAliveRequest request = KeepAliveRequest.builder()
        .withSessionIds(sessionIds)
        .withCommandSequences(commandResponses)
        .withEventIndexes(eventIndexes)
        .build();

    long keepAliveTime = System.currentTimeMillis();
    connection.keepAlive(request).whenComplete((response, error) -> {
      if (open.get()) {
        long delta = System.currentTimeMillis() - keepAliveTime;
        if (error == null) {
          // If the request was successful, update the address selector and schedule the next keep-alive.
          if (response.status() == RaftResponse.Status.OK) {
            selectorManager.resetAll(response.leader(), response.members());

            // Iterate through sessions and close sessions that weren't kept alive by the request (have already been closed).
            Set<Long> keptAliveSessions = Sets.newHashSet(Longs.asList(response.sessionIds()));
            for (RaftSessionState session : needKeepAlive) {
              if (keptAliveSessions.contains(session.getSessionId().id())) {
                session.setState(PrimitiveState.CONNECTED);
              } else {
                session.setState(PrimitiveState.EXPIRED);
              }
            }
            scheduleKeepAlive(System.currentTimeMillis(), sessionTimeout, delta);
          }
          // If the timeout has not been passed, attempt to keep the session alive again with no delay.
          // We will continue to retry until the session expiration has passed.
          else if (System.currentTimeMillis() - lastKeepAliveTime < sessionTimeout) {
            selectorManager.resetAll(null, connection.members());
            keepAliveSessions(lastKeepAliveTime, sessionTimeout);
          }
          // If no leader was set, set the session state to unstable and schedule another keep-alive.
          else {
            needKeepAlive.forEach(s -> s.setState(PrimitiveState.SUSPENDED));
            selectorManager.resetAll();
            scheduleKeepAlive(lastKeepAliveTime, sessionTimeout, delta);
          }
        }
        // If the timeout has not been passed, reset the connection and attempt to keep the session alive
        // again with no delay.
        else if (System.currentTimeMillis() - lastKeepAliveTime < sessionTimeout && connection.leader() != null) {
          selectorManager.resetAll(null, connection.members());
          keepAliveSessions(lastKeepAliveTime, sessionTimeout);
        }
        // If no leader was set, set the session state to unstable and schedule another keep-alive.
        else {
          needKeepAlive.forEach(s -> s.setState(PrimitiveState.SUSPENDED));
          selectorManager.resetAll();
          scheduleKeepAlive(lastKeepAliveTime, sessionTimeout, delta);
        }
      }
    });
  }
1095350
async def _handle_rtp_packet(self, packet: RtpPacket, arrival_time_ms: int):
        
        self.__log_debug('< %s', packet)

        # feed bitrate estimator
        if self.__remote_bitrate_estimator is not None:
            if packet.extensions.abs_send_time is not None:
                remb = self.__remote_bitrate_estimator.add(
                    abs_send_time=packet.extensions.abs_send_time,
                    arrival_time_ms=arrival_time_ms,
                    payload_size=len(packet.payload) + packet.padding_size,
                    ssrc=packet.ssrc,
                )
                if self.__rtcp_ssrc is not None and remb is not None:
                    # send Receiver Estimated Maximum Bitrate feedback
                    rtcp_packet = RtcpPsfbPacket(
                        fmt=RTCP_PSFB_APP, ssrc=self.__rtcp_ssrc, media_ssrc=0)
                    rtcp_packet.fci = pack_remb_fci(*remb)
                    await self._send_rtcp(rtcp_packet)

        # keep track of sources
        self.__active_ssrc[packet.ssrc] = clock.current_datetime()

        # check the codec is known
        codec = self.__codecs.get(packet.payload_type)
        if codec is None:
            self.__log_debug('x RTP packet with unknown payload type %d', packet.payload_type)
            return

        # feed RTCP statistics
        if packet.ssrc not in self.__remote_streams:
            self.__remote_streams[packet.ssrc] = StreamStatistics(codec.clockRate)
        self.__remote_streams[packet.ssrc].add(packet)

        # unwrap retransmission packet
        if is_rtx(codec):
            original_ssrc = self.__rtx_ssrc.get(packet.ssrc)
            if original_ssrc is None:
                self.__log_debug('x RTX packet from unknown SSRC %d', packet.ssrc)
                return

            if len(packet.payload) < 2:
                return

            codec = self.__codecs[codec.parameters['apt']]
            packet = unwrap_rtx(packet,
                                payload_type=codec.payloadType,
                                ssrc=original_ssrc)

        # send NACKs for any missing any packets
        if self.__nack_generator is not None and self.__nack_generator.add(packet):
            await self._send_rtcp_nack(packet.ssrc, sorted(self.__nack_generator.missing))

        # parse codec-specific information
        try:
            if packet.payload:
                packet._data = depayload(codec, packet.payload)
            else:
                packet._data = b''
        except ValueError as exc:
            self.__log_debug('x RTP payload parsing failed: %s', exc)
            return

        # try to re-assemble encoded frame
        encoded_frame = self.__jitter_buffer.add(packet)

        # if we have a complete encoded frame, decode it
        if encoded_frame is not None and self.__decoder_thread:
            encoded_frame.timestamp = self.__timestamp_mapper.map(encoded_frame.timestamp)
            self.__decoder_queue.put((codec, encoded_frame))
98656
def PLAY(self):
        """"""
        message = "PLAY " + self.session.url + " RTSP/1.0\r\n"
        message += self.sequence
        message += self.authentication
        message += self.user_agent
        message += self.session_id
        message += '\r\n'
        return message
431639
def ReadResponsesForRequestId(self, session_id, request_id, timestamp=None):
    
    request = rdf_flow_runner.RequestState(id=request_id, session_id=session_id)
    for _, responses in self.ReadResponses([request], timestamp=timestamp):
      return responses
90932
protected void handleInvocationResponse (int reqId, int methodId, Object[] args)
    {
        // look up the invocation marshaller registered for that response
        ListenerMarshaller listener = _listeners.remove(reqId);
        if (listener == null) {
            log.warning("Received invocation response for which we have no registered listener. " +
                        "It is possible that this listener was flushed because the response did " +
                        "not arrive within " + _listenerMaxAge + " milliseconds.",
                        "reqId", reqId, "methId", methodId, "args", args);
            return;
        }

//         log.info("Dispatching invocation response", "listener", listener,
//                  "methId", methodId, "args", args);

        // dispatch the response
        try {
            listener.dispatchResponse(methodId, args);
        } catch (Throwable t) {
            log.warning("Invocation response listener choked", "listener", listener,
                        "methId", methodId, "args", args, t);
        }

        // flush expired listeners periodically
        long now = System.currentTimeMillis();
        if (now - _lastFlushTime > LISTENER_FLUSH_INTERVAL) {
            _lastFlushTime = now;
            flushListeners(now);
        }
    }
1170516
def parseRtspReply(self, data):
    print("Parsing Received Rtsp data...")

    """"""
    lines = data.split('\n')
    seqNum = int(lines[1].split(' ')[1])

    # Process only if the server reply's sequence number is the same as the request's
    if seqNum == self.rtspSeq:
      session = int(lines[2].split(' ')[1])
      # New RTSP session ID
      if self.sessionId == 0:
        self.sessionId = session

      # Process only if the session ID is the same
      if self.sessionId == session:
        if int(lines[0].split(' ')[1]) == 200:
          if self.requestSent == self.SETUP:
            #-------------
            # TO COMPLETE
            #-------------
            # Update RTSP state.
            print("Updating RTSP state...")
            # self.state = ...
            self.state = self.READY
            # Open RTP port.
            #self.openRtpPort()
            print("Setting Up RtpPort for Video Stream")
            self.openRtpPort()

          elif self.requestSent == self.PLAY:
             self.state = self.PLAYING
             print('-'*60 + "\nClient is PLAYING...\n" + '-'*60)
          elif self.requestSent == self.PAUSE:
             self.state = self.READY

            # The play thread exits. A new thread is created on resume.
             self.playEvent.set()

          elif self.requestSent == self.TEARDOWN:
            # self.state = ...

            # Flag the teardownAcked to close the socket.
            self.teardownAcked = 1
438376
def handle_response(response)
      if response.is_a?(Result)
        token = response.token
        if (result = OperationResult.from_results(response))
          if result.non_delivery?
            @non_delivery_stats.update(result.content.nil? ? "nil" : result.content)
          elsif result.error?
            @result_error_stats.update(result.content.nil? ? "nil" : result.content)
          end
          @result_stats.update(result.status)
        else
          @result_stats.update(response.results.nil? ? "nil" : response.results)
        end

        if (pending_request = @pending_requests[token])
          if result && result.non_delivery? && pending_request.kind == :send_request
            if result.content == OperationResult::TARGET_NOT_CONNECTED
              # Log and temporarily ignore so that timeout retry mechanism continues, but save reason for use below if timeout
              # Leave purging of associated request until final response, i.e., success response or retry timeout
              if (parent_token = pending_request.retry_parent_token)
                @pending_requests[parent_token].non_delivery = result.content
              else
                pending_request.non_delivery = result.content
              end
              Log.info("Non-delivery of <#{token}> because #{result.content}")
            elsif result.content == OperationResult::RETRY_TIMEOUT && pending_request.non_delivery
              # Request timed out but due to another non-delivery reason, so use that reason since more germane
              response.results = OperationResult.non_delivery(pending_request.non_delivery)
              deliver_response(response, pending_request)
            else
              deliver_response(response, pending_request)
            end
          else
            deliver_response(response, pending_request)
          end
        elsif result && result.non_delivery?
          Log.info("Non-delivery of <#{token}> because #{result.content}")
        else
          Log.debug("No pending request for response #{response.to_s([])}")
        end
      end
      true
    end
492289
def timeout_callback(self):
        
        # Check if elapsed time since last response exceeds our configured
        # maximum keep alive timeout value
        now = time()
        time_elapsed = now - self._last_response_time
        if time_elapsed < self.timeout:
            time_left = self.timeout - time_elapsed
            self._timeout_handler = (
                self._loop.call_later(
                    time_left,
                    self.timeout_callback
                )
            )
        else:
            logger.info('KeepAlive Timeout. Closing connection.')
            responseb = self.encoder({
                "MPRPC": self.VERSION,
                "CODE": 504
            })
            self._stream_writer.write(responseb)
            self.close()
1189808
def _respond(self, resp):
        """"""
        response_queue = self._response_queues.get(timeout=0.1)
        response_queue.put(resp)
        self._completed_response_lines = []
        self._is_multiline = None
153809
async def process_response(self, request, response):
        
        if _REISSUE_KEY in request:
            if (response.started or
                not isinstance(response, web.Response) or
                response.status < 200 or response.status > 299):
                return

            await self.remember_ticket(request, request[_REISSUE_KEY])
754647
protected int checkResponse(TCPReadRequestContext rsc) {
        // Parse the proxy server response
        //
        WsByteBuffer[] buffers = rsc.getBuffers();

        // check if the correct response was received
        if (null == buffers) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "Could not complete proxy handshake, null buffers");
            }
            return STATUS_ERROR;
        }
        int status = validateProxyResponse(buffers);
        if (STATUS_DONE == status) {
            releaseProxyReadBuffer();
        }
        return status;
    }
1327590
@Override
    public void handle(T result, HttpResponse response) {
        String transitionHeader = response.getHeaders().get(Headers.IBM_TRANSITION);
        if (transitionHeader != null) {
            result.setTransitionDate(parseDate(transitionHeader));
            String transition = parseTransition(transitionHeader);
            if (transition != null) {
                result.setTransition(transition);
            }
        }
    }