+/**
+ * @brief Send message to service that this client does not want to receive
+ * further updates from the biased peer stream
+ *
+ * @param rps_handle The handle representing the service to the client
+ */
+static void
+cancel_stream (struct GNUNET_RPS_Handle *rps_handle)
+{
+ struct GNUNET_MQ_Envelope *ev;
+
+ ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_CS_DEBUG_STREAM_CANCEL);
+ GNUNET_MQ_send (rps_handle->mq, ev);
+}
+
+
+/**
+ * @brief Cancel a specific request for updates from the biased peer stream
+ *
+ * @param srh The request handle to cancel
+ */
+void
+GNUNET_RPS_stream_cancel (struct GNUNET_RPS_StreamRequestHandle *srh)
+{
+ struct GNUNET_RPS_Handle *rps_handle;
+
+ rps_handle = srh->rps_handle;
+ GNUNET_CONTAINER_DLL_remove (rps_handle->stream_requests_head,
+ rps_handle->stream_requests_tail,
+ srh);
+ GNUNET_free (srh);
+ if (NULL == rps_handle->stream_requests_head) cancel_stream (rps_handle);
+}
+
+
+/**
+ * This function is called, when the service sends another peer from the biased
+ * stream.
+ * It calls the callback the caller provided
+ * and disconnects afterwards.
+ *
+ * TODO merge with check_view_update
+ *
+ * @param msg the message
+ */
+static int
+check_stream_input (void *cls,
+ const struct GNUNET_RPS_CS_DEBUG_StreamReply *msg)
+{
+ uint16_t msize = ntohs (msg->header.size);
+ uint32_t num_peers = ntohl (msg->num_peers);
+ (void) cls;
+
+ msize -= sizeof (struct GNUNET_RPS_CS_DEBUG_StreamReply);
+ if ( (msize / sizeof (struct GNUNET_PeerIdentity) != num_peers) ||
+ (msize % sizeof (struct GNUNET_PeerIdentity) != 0) )
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ return GNUNET_OK;
+}
+
+/**
+ * This function is called, when the service sends another peer from the biased
+ * stream.
+ * It calls the callback the caller provided
+ * and disconnects afterwards.
+ *
+ * @param msg the message
+ */
+static void
+handle_stream_input (void *cls,
+ const struct GNUNET_RPS_CS_DEBUG_StreamReply *msg)
+{
+ struct GNUNET_RPS_Handle *h = cls;
+ const struct GNUNET_PeerIdentity *peers;
+ /* The following two pointers are used to prevent that new handles are
+ * inserted into the DLL, that is currently iterated over, from within a call
+ * to that handler_cb, are executed and in turn again add themselves to the
+ * iterated DLL infinitely */
+ struct GNUNET_RPS_StreamRequestHandle *srh_head_tmp;
+ struct GNUNET_RPS_StreamRequestHandle *srh_tail_tmp;
+ uint64_t num_peers;
+ uint64_t num_peers_return;
+
+ peers = (struct GNUNET_PeerIdentity *) &msg[1];
+ num_peers = ntohl (msg->num_peers);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Received %" PRIu64 " peer(s) from stream input.\n",
+ num_peers);
+ srh_head_tmp = h->stream_requests_head;
+ srh_tail_tmp = h->stream_requests_tail;
+ h->stream_requests_head = NULL;
+ h->stream_requests_tail = NULL;
+ for (struct GNUNET_RPS_StreamRequestHandle *srh_iter = srh_head_tmp;
+ NULL != srh_iter;
+ srh_iter = srh_iter->next)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Calling srh - left: %" PRIu64 "\n",
+ srh_iter->num_peers_left);
+ if (0 == srh_iter->num_peers_left) /* infinite updates */
+ {
+ num_peers_return = num_peers;
+ }
+ else if (num_peers > srh_iter->num_peers_left)
+ {
+ num_peers_return = num_peers - srh_iter->num_peers_left;
+ }
+ else /* num_peers <= srh_iter->num_peers_left */
+ {
+ num_peers_return = srh_iter->num_peers_left - num_peers;
+ }
+ srh_iter->ready_cb (srh_iter->ready_cb_cls,
+ num_peers_return,
+ peers);
+ if (0 == srh_iter->num_peers_left) ;
+ else if (num_peers_return >= srh_iter->num_peers_left)
+ {
+ remove_stream_request (srh_iter,
+ srh_head_tmp,
+ srh_tail_tmp);
+ }
+ else
+ {
+ srh_iter->num_peers_left -= num_peers_return;
+ }
+ }
+ for (struct GNUNET_RPS_StreamRequestHandle *srh_iter = srh_head_tmp;
+ NULL != srh_iter;
+ srh_iter = srh_iter->next)
+ {
+ GNUNET_CONTAINER_DLL_insert (h->stream_requests_head,
+ h->stream_requests_tail,
+ srh_iter);
+ }
+
+ if (NULL == h->stream_requests_head)
+ {
+ cancel_stream (h);
+ }
+}
+