on_response_output (GObject *pollable,
                    gpointer user_data)
{
  CockpitWebResponse *self = user_data;
  GError *error = NULL;
  const guint8 *data;
  GBytes *block;
  gssize count;
  gsize before, size, len;

  block = g_queue_peek_head (self->queue);
  if (block)
    {
      data = g_bytes_get_data (block, &len);
      g_assert (len == 0 || self->partial_offset < len);
      data += self->partial_offset;
      len -= self->partial_offset;

      before = self->out_queued;

      if (len > 0)
        {
          count = g_pollable_output_stream_write_nonblocking (self->out, data, len,
                                                              NULL, &error);
        }
      else
        {
          count = 0;
        }

      if (count < 0)
        {
          if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
            {
              g_error_free (error);
              return TRUE;
            }

          if (!cockpit_web_should_suppress_output_error (self->logname, error))
            g_message ("%s: couldn't write web output: %s", self->logname, error->message);

          self->failed = TRUE;
          cockpit_web_response_done (self);

          g_error_free (error);
          return FALSE;
        }

      if (count == len)
        {
          g_debug ("%s: sent %d bytes", self->logname, (int)len);
          self->partial_offset = 0;
          block = g_queue_pop_head (self->queue);
          size = g_bytes_get_size (block);
          g_assert (size <= self->out_queued);
          self->out_queued -= size;
          g_bytes_unref (block);
        }
      else
        {
          g_debug ("%s: sent %d partial", self->logname, (int)count);
          g_assert (count < len);
          self->partial_offset += count;
        }

      /*
       * If we're controlling another flow, turn it on again when our output
       * buffer size becomes less than the low mark.
       */
      if (before >= QUEUE_PRESSURE && self->out_queued < QUEUE_PRESSURE)
        cockpit_flow_emit_pressure (COCKPIT_FLOW (self), FALSE);

      return TRUE;
    }
  else
    {
      g_source_destroy (self->source);
      g_source_unref (self->source);
      self->source = NULL;

      if (self->complete)
        {
          g_debug ("%s: complete flushing output", self->logname);
          g_output_stream_flush_async (G_OUTPUT_STREAM (self->out), G_PRIORITY_DEFAULT,
                                       NULL, on_output_flushed, g_object_ref (self));
        }

      return FALSE;
    }
}
on_response_output (GObject *pollable,
                    gpointer user_data)
{
  CockpitWebResponse *self = user_data;
  GError *error = NULL;
  const guint8 *data;
  GBytes *block;
  gssize count;
  gsize before, size, len;

  block = g_queue_peek_head (self->queue);
  if (block)
    {
      data = g_bytes_get_data (block, &len);
      g_assert (len == 0 || self->partial_offset < len);
      data += self->partial_offset;
      len -= self->partial_offset;

      before = self->out_queued;

      if (len > 0)
        {
          count = g_pollable_output_stream_write_nonblocking (self->out, data, len,
                                                              NULL, &error);
        }
      else
        {
          count = 0;
        }

      if (count < 0)
        {
          if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
            {
              g_error_free (error);
              return TRUE;
            }

          if (!cockpit_web_should_suppress_output_error (self->logname, error))
            g_message ("%s: couldn't write web output: %s", self->logname, error->message);

          self->failed = TRUE;
          cockpit_web_response_done (self);

          g_error_free (error);
          return FALSE;
        }

      if (count == len)
        {
          g_debug ("%s: sent %d bytes", self->logname, (int)len);
          self->partial_offset = 0;
          block = g_queue_pop_head (self->queue);
          size = g_bytes_get_size (block);
          g_assert (size <= self->out_queued);
          self->out_queued -= size;
          g_bytes_unref (block);
        }
      else
        {
          g_debug ("%s: sent %d partial", self->logname, (int)count);
          g_assert (count < len);
          self->partial_offset += count;
        }

       
      if (before >= QUEUE_PRESSURE && self->out_queued < QUEUE_PRESSURE)
        cockpit_flow_emit_pressure (COCKPIT_FLOW (self), FALSE);

      return TRUE;
    }
  else
    {
      g_source_destroy (self->source);
      g_source_unref (self->source);
      self->source = NULL;

      if (self->complete)
        {
          g_debug ("%s: complete flushing output", self->logname);
          g_output_stream_flush_async (G_OUTPUT_STREAM (self->out), G_PRIORITY_DEFAULT,
                                       NULL, on_output_flushed, g_object_ref (self));
        }

      return FALSE;
    }
}