<commit_before>/*
 * update.c :  entry point for update RA functions for ra_serf
 *
 * ====================================================================
 * Copyright (c) 2006 CollabNet.  All rights reserved.
 *
 * This software is licensed as described in the file COPYING, which
 * you should have received as part of this distribution.  The terms
 * are also available at http://subversion.tigris.org/license-1.html.
 * If newer versions of this license are posted there, you may use a
 * newer version instead, at your option.
 *
 * This software consists of voluntary contributions made by many
 * individuals.  For exact contribution history, see the revision
 * history and logs, available at http://subversion.tigris.org/.
 * ====================================================================
 */




#define APR_WANT_STRFUNC
#include <apr_want.h>

#include <apr_uri.h>

#include <expat.h>

#include <serf.h>

#include "svn_pools.h"
#include "svn_ra.h"
#include "svn_dav.h"
#include "svn_xml.h"
#include "../libsvn_ra/ra_loader.h"
#include "svn_config.h"
#include "svn_delta.h"
#include "svn_version.h"
#include "svn_path.h"
#include "svn_base64.h"
#include "svn_private_config.h"

#include "ra_serf.h"



/*
 * This enum represents the current state of our XML parsing for a REPORT.
 */
typedef enum {
    OPEN_DIR,
    ADD_DIR,
    OPEN_FILE,
    ADD_FILE,
    PROP,
    IGNORE_PROP_NAME,
    NEED_PROP_NAME,
} report_state_e;

/*
 * This structure represents the information for a directory.
 */
typedef struct report_dir_t
{
  /* Our parent directory.
   *
   * This value is NULL when we are the root.
   */
  struct report_dir_t *parent_dir;

  apr_pool_t *pool;

  /* Our name sans any parents. */
  const char *base_name;

  /* the expanded directory name (including all parent names) */
  const char *name;

  /* temporary path buffer for this directory. */
  svn_stringbuf_t *name_buf;

  /* the canonical url for this directory. */
  const char *url;

  /* Our base revision - SVN_INVALID_REVNUM if we're adding this dir. */
  svn_revnum_t base_rev;

  /* The target revision we're retrieving. */
  svn_revnum_t target_rev;

  /* controlling dir baton - this is only created in open_dir() */
  void *dir_baton;
  apr_pool_t *dir_baton_pool;

  /* Our master update editor and baton. */
  const svn_delta_editor_t *update_editor;
  void *update_baton;

  /* How many references to this directory do we still have open? */
  apr_size_t ref_count;

  /* Namespace list allocated out of this ->pool. */
  svn_ra_serf__ns_t *ns_list;

  /* hashtable for all of the properties (shared within a dir) */
  apr_hash_t *props;

  /* hashtable for all to-be-removed properties (shared within a dir) */
  apr_hash_t *removed_props;

  /* The propfind request for our current directory */
  svn_ra_serf__propfind_context_t *propfind;

  /* Has the server told us to fetch the dir props? */
  svn_boolean_t fetch_props;

  /* Have we closed the directory tag (meaning no more additions)? */
  svn_boolean_t tag_closed;

  /* The children of this directory  */
  struct report_dir_t *children;

  /* The next sibling of this directory */
  struct report_dir_t *sibling;
} report_dir_t;

/*
 * This structure represents the information for a file.
 *
 * A directory may have a report_info_t associated with it as well.
 *
 * This structure is created as we parse the REPORT response and
 * once the element is completed, we create a report_fetch_t structure
 * to give to serf to retrieve this file.
 */
typedef struct report_info_t
{
  apr_pool_t *pool;

  /* The enclosing directory.
   *
   * If this structure refers to a directory, the dir it points to will be
   * itself.
   */
  report_dir_t *dir;

  /* Our name sans any directory info. */
  const char *base_name;

  /* the expanded file name (including all parent directory names) */
  const char *name;

  /* file name buffer */
  svn_stringbuf_t *name_buf;

  /* the canonical url for this file. */
  const char *url;

  /* Our base revision - SVN_INVALID_REVNUM if we're adding this file. */
  svn_revnum_t base_rev;

  /* The target revision we're retrieving. */
  svn_revnum_t target_rev;

  /* our delta base, if present (NULL if we're adding the file) */
  const svn_string_t *delta_base;

  /* The propfind request for our current file (if present) */
  svn_ra_serf__propfind_context_t *propfind;

  /* Has the server told us to fetch the file props? */
  svn_boolean_t fetch_props;

  /* Has the server told us to go fetch - only valid if we had it already */
  svn_boolean_t fetch_file;

  /* pool passed to update->add_file, etc. */
  apr_pool_t *editor_pool;

  /* controlling file_baton and textdelta handler */
  void *file_baton;
  svn_txdelta_window_handler_t textdelta;
  void *textdelta_baton;

  /* temporary property for this file which is currently being parsed
   * It will eventually be stored in our parent directory's property hash.
   */
  const char *prop_ns;
  const char *prop_name;
  const char *prop_val;
  apr_size_t prop_val_len;
  const char *prop_encoding;
} report_info_t;

/*
 * This structure represents a single request to GET (fetch) a file with
 * its associated Serf session/connection.
 */
typedef struct report_fetch_t {
  /* Our pool. */
  apr_pool_t *pool;

  /* Non-NULL if we received an error during processing. */
  svn_error_t *err;

  /* The session we should use to fetch the file. */
  svn_ra_serf__session_t *sess;

  /* The connection we should use to fetch file. */
  svn_ra_serf__connection_t *conn;

  /* Stores the information for the file we want to fetch. */
  report_info_t *info;

  /* Have we read our response headers yet? */
  svn_boolean_t read_headers;

  /* This flag is set when our response is aborted before we reach the
   * end and we decide to requeue this request.
   */
  svn_boolean_t aborted_read;
  apr_off_t aborted_read_size;

  /* This is the amount of data that we have read so far. */
  apr_off_t read_size;

  /* If we're receiving an svndiff, this will be non-NULL. */
  svn_stream_t *delta_stream;

  /* If we're writing this file to a stream, this will be non-NULL. */
  svn_stream_t *target_stream;

  /* Are we done fetching this file? */
  svn_boolean_t done;
  svn_ra_serf__list_t **done_list;
  svn_ra_serf__list_t done_item;

} report_fetch_t;

/*
 * Encapsulates all of the REPORT parsing state that we need to know at
 * any given time.
 *
 * Previous states are stored in ->prev field.
 */
typedef struct report_state_list_t {
   /* The current state that we are in now. */
  report_state_e state;

  /* Information */
  report_info_t *info;

  /* Temporary pool */
  apr_pool_t *pool;

  /* Temporary namespace list allocated from ->pool */
  svn_ra_serf__ns_t *ns_list;

  /* The previous state we were in. */
  struct report_state_list_t *prev;
} report_state_list_t;

/*
 * The master structure for a REPORT request and response.
 */
typedef struct {
  apr_pool_t *pool;

  svn_ra_serf__session_t *sess;
  svn_ra_serf__connection_t *conn;

  /* Source path and destination path */
  const char *source;
  const char *destination;

  /* Our update target. */
  const char *update_target;

  /* What is the target revision that we want for this REPORT? */
  svn_revnum_t target_rev;

  /* Have we been asked to ignore ancestry, recursion, or textdeltas? */
  svn_boolean_t ignore_ancestry;
  svn_boolean_t recurse;
  svn_boolean_t text_deltas;

  /* Our master update editor and baton. */
  const svn_delta_editor_t *update_editor;
  void *update_baton;

  /* The request body for the REPORT. */
  serf_bucket_t *buckets;

  /* Our XML parser and root namespace for parsing the response. */
  XML_Parser xmlp;
  svn_ra_serf__ns_t *ns_list;

  /* the current state we are in for parsing the REPORT response.
   *
   * could allocate this as an array rather than a linked list.
   *
   * (We tend to use only about 8 or 9 states in a given update-report,
   * but in theory it could be much larger based on the number of directories
   * we are adding.)
   */
  report_state_list_t *state;
  /* A list of previous states that we have created but aren't using now. */
  report_state_list_t *free_state;

  /* root directory object */
  report_dir_t *root_dir;

  /* number of pending GET requests */
  unsigned int active_fetches;

  /* completed fetches (contains report_fetch_t) */
  svn_ra_serf__list_t *done_fetches;

  /* number of pending PROPFIND requests */
  unsigned int active_propfinds;

  /* completed PROPFIND requests (contains propfind_context_t) */
  svn_ra_serf__list_t *done_propfinds;

  /* list of files that will only have prop changes (contains report_info_t) */
  svn_ra_serf__list_t *file_propchanges_only;

  /* free list of info structures */
  report_info_t *free_info;

  /* The path to the REPORT request */
  const char *path;

  /* Are we done parsing the REPORT response? */
  svn_boolean_t done;

} report_context_t;



static void push_state(report_context_t *ctx, report_state_e state)
{
  report_state_list_t *new_state;

  if (!ctx->free_state)
    {
      new_state = apr_palloc(ctx->sess->pool, sizeof(*ctx->state));

      apr_pool_create(&new_state->pool, ctx->sess->pool);
    }
  else
    {
      new_state = ctx->free_state;
      ctx->free_state = ctx->free_state->prev;

      apr_pool_clear(new_state->pool);
    }
  new_state->state = state;

  if (!ctx->state && state == OPEN_DIR)
    {
      new_state->info = apr_palloc(ctx->sess->pool, sizeof(*new_state->info));
      apr_pool_create(&new_state->info->pool, ctx->sess->pool);

      /* Create our root state now. */
      new_state->info->dir = apr_pcalloc(ctx->sess->pool,
                                        sizeof(*new_state->info->dir));
      new_state->info->dir->pool = new_state->info->pool;

      /* Create the root property tree. */
      new_state->info->dir->props = apr_hash_make(new_state->info->pool);
      new_state->info->dir->removed_props =
          apr_hash_make(new_state->info->pool);

      /* Point to the update_editor */
      new_state->info->dir->update_editor = ctx->update_editor;
      new_state->info->dir->update_baton = ctx->update_baton;

      /* Allow us to be found later. */
      ctx->root_dir = new_state->info->dir;
    }
  else if (state == ADD_DIR || state == OPEN_DIR)
    {
      new_state->info = apr_palloc(ctx->state->info->pool,
                                   sizeof(*new_state->info));
      apr_pool_create(&new_state->info->pool, ctx->state->info->pool);

      new_state->info->dir =
          apr_pcalloc(ctx->state->info->pool, sizeof(*new_state->info->dir));
      new_state->info->dir->pool = new_state->info->pool;
      new_state->info->dir->parent_dir = ctx->state->info->dir;
      new_state->info->dir->parent_dir->ref_count++;

      new_state->info->dir->props = apr_hash_make(new_state->info->pool);
      new_state->info->dir->removed_props =
          apr_hash_make(new_state->info->pool);

      /* Point our ns_list at our parents to try to reuse it. */
      new_state->info->dir->ns_list =
          new_state->info->dir->parent_dir->ns_list;

      /* Point to the update_editor */
      new_state->info->dir->update_editor = ctx->update_editor;
      new_state->info->dir->update_baton = ctx->update_baton;

      /* Add ourselves to our parent's list */
      new_state->info->dir->sibling = ctx->state->info->dir->children;
      ctx->state->info->dir->children = new_state->info->dir;
    }
  else if (state == OPEN_FILE || state == ADD_FILE)
    {
      new_state->info = apr_palloc(ctx->state->info->pool,
                                   sizeof(*new_state->info));
      apr_pool_create(&new_state->info->pool, ctx->state->info->pool);
      new_state->info->file_baton = NULL;

      /* Point at our parent's directory state. */
      new_state->info->dir = ctx->state->info->dir;
      new_state->info->dir->ref_count++;
    }
  /* if we have state info from our parent, reuse it. */
  else if (ctx->state && ctx->state->info)
    {
      new_state->info = ctx->state->info;
    }
  else
    {
      abort();
    }

  if (!ctx->state)
    {
      /* Attach to the root state. */
      new_state->ns_list = ctx->ns_list;
    }
  else
    {
      new_state->ns_list = ctx->state->ns_list;
    }

  /* Add it to the state chain. */
  new_state->prev = ctx->state;
  ctx->state = new_state;
}

static void pop_state(report_context_t *ctx)
{
  report_state_list_t *free_state;
  free_state = ctx->state;
  /* advance the current state */
  ctx->state = ctx->state->prev;
  free_state->prev = ctx->free_state;
  ctx->free_state = free_state;
  ctx->free_state->info = NULL;
}

static svn_error_t *
set_file_props(void *baton,
               const char *ns, apr_ssize_t ns_len,
               const char *name, apr_ssize_t name_len,
               const svn_string_t *val,
               apr_pool_t *pool)
{
  report_info_t *info = baton;
  const svn_delta_editor_t *editor = info->dir->update_editor;

  return svn_ra_serf__set_baton_props(editor->change_file_prop,
                                      info->file_baton,
                                      ns, ns_len, name, name_len, val, pool);
}

static svn_error_t *
set_dir_props(void *baton,
              const char *ns, apr_ssize_t ns_len,
              const char *name, apr_ssize_t name_len,
              const svn_string_t *val,
              apr_pool_t *pool)
{
  report_dir_t *dir = baton;
  return svn_ra_serf__set_baton_props(dir->update_editor->change_dir_prop,
                                      dir->dir_baton,
                                      ns, ns_len, name, name_len, val, pool);
}

static svn_error_t *
remove_file_props(void *baton,
                  const char *ns, apr_ssize_t ns_len,
                  const char *name, apr_ssize_t name_len,
                  const svn_string_t *val,
                  apr_pool_t *pool)
{
  report_info_t *info = baton;
  const svn_delta_editor_t *editor = info->dir->update_editor;

  return svn_ra_serf__set_baton_props(editor->change_file_prop,
                                      info->file_baton,
                                      ns, ns_len, name, name_len, NULL, pool);
}

static svn_error_t *
remove_dir_props(void *baton,
                 const char *ns, apr_ssize_t ns_len,
                 const char *name, apr_ssize_t name_len,
                 const svn_string_t *val,
                 apr_pool_t *pool)
{
  report_dir_t *dir = baton;
  svn_ra_serf__set_baton_props(dir->update_editor->change_dir_prop,
                               dir->dir_baton,
                               ns, ns_len, name, name_len, NULL, pool);
}

static svn_error_t*
open_dir(report_dir_t *dir)
{
  /* if we're already open, return now */
  if (dir->dir_baton)
    {
      return SVN_NO_ERROR;
    }

  if (dir->base_name[0] == '\0')
    {
      apr_pool_create(&dir->dir_baton_pool, dir->pool);

      dir->name_buf = svn_stringbuf_create("", dir->dir_baton_pool);
      dir->name = dir->name_buf->data;

      SVN_ERR(dir->update_editor->open_root(dir->update_baton, dir->base_rev,
                                            dir->dir_baton_pool,
                                            &dir->dir_baton));
    }
  else
    {
      SVN_ERR(open_dir(dir->parent_dir));

      apr_pool_create(&dir->dir_baton_pool, dir->parent_dir->dir_baton_pool);

      /* Expand our name. */
      dir->name_buf = svn_stringbuf_dup(dir->parent_dir->name_buf,
                                        dir->dir_baton_pool);
      svn_path_add_component(dir->name_buf, dir->base_name);
      dir->name = dir->name_buf->data;

      if (SVN_IS_VALID_REVNUM(dir->base_rev))
        {
          SVN_ERR(dir->update_editor->open_directory(dir->name,
                                         dir->parent_dir->dir_baton,
                                         dir->base_rev, dir->dir_baton_pool,
                                         &dir->dir_baton));
        }
      else
        {
          SVN_ERR(dir->update_editor->add_directory(dir->name,
                                                    dir->parent_dir->dir_baton,
                                                    NULL, SVN_INVALID_REVNUM,
                                                    dir->dir_baton_pool,
                                                    &dir->dir_baton));
        }
    }

  return SVN_NO_ERROR;
}

static svn_error_t *
close_dir(report_dir_t *dir)
{
  report_dir_t *prev, *sibling;

  if (dir->ref_count)
    {
      abort();
    }

  svn_ra_serf__walk_all_props(dir->props, dir->base_name, dir->base_rev,
                              set_dir_props, dir, dir->dir_baton_pool);

  svn_ra_serf__walk_all_props(dir->removed_props, dir->base_name,
                              dir->base_rev, remove_dir_props, dir,
                              dir->dir_baton_pool);

  if (dir->fetch_props)
    {
      svn_ra_serf__walk_all_props(dir->props, dir->url, dir->target_rev,
                                  set_dir_props, dir, dir->dir_baton_pool);
    }

  SVN_ERR(dir->update_editor->close_directory(dir->dir_baton,
                                              dir->dir_baton_pool));

  /* remove us from our parent's children list */
  if (dir->parent_dir)
    {
      prev = NULL;
      sibling = dir->parent_dir->children;

      while (sibling != dir)
        {
          prev = sibling;
          sibling = sibling->sibling;
          if (!sibling)
            abort();
        }

      if (!prev)
        {
          dir->parent_dir->children = dir->sibling;
        }
      else
        {
          prev->sibling = dir->sibling;
        }
    }

  apr_pool_destroy(dir->dir_baton_pool);
  apr_pool_destroy(dir->pool);

  return SVN_NO_ERROR;
}

static svn_error_t *close_all_dirs(report_dir_t *dir)
{
  while (dir->children)
    {
      SVN_ERR(close_all_dirs(dir->children));
      dir->ref_count--;
    }

  if (dir->ref_count)
    {
      abort();
    }

  if (!dir->dir_baton)
    {
      SVN_ERR(open_dir(dir));
    }

  return close_dir(dir);
}

static apr_status_t
headers_fetch(serf_bucket_t *headers,
              void *baton,
              apr_pool_t *pool)
{
  report_fetch_t *fetch_ctx = baton;

  /* note that we have old VC URL */
  if (SVN_IS_VALID_REVNUM(fetch_ctx->info->base_rev))
    {
      serf_bucket_headers_setn(headers, SVN_DAV_DELTA_BASE_HEADER,
                               fetch_ctx->info->delta_base->data);
      serf_bucket_headers_setn(headers, "Accept-Encoding",
                               "svndiff1;q=0.9,svndiff;q=0.8");
    }
  else
    {
      serf_bucket_headers_setn(headers, "Accept-Encoding", "gzip");
    }

  return APR_SUCCESS;
}

static apr_status_t
cancel_fetch(serf_request_t *request,
             serf_bucket_t *response,
             int status_code,
             void *baton)
{
  report_fetch_t *fetch_ctx = baton;

  /* Uh-oh.  Our connection died on us.
   *
   * The core ra_serf layer will requeue our request - we just need to note
   * that we got cut off in the middle of our song.
   */
  if (!response)
    {
      /* If we already started the fetch and opened the file handle, we need
       * to hold subsequent read() ops until we get back to where we were
       * before the close and we can then resume the textdelta() calls.
       */
      if (fetch_ctx->read_headers == TRUE)
        {
          if (fetch_ctx->aborted_read == FALSE && fetch_ctx->read_size)
            {
              fetch_ctx->aborted_read = TRUE;
              fetch_ctx->aborted_read_size = fetch_ctx->read_size;
            }
          fetch_ctx->read_size = 0;
        }

      return APR_SUCCESS;
    }

  /* We have no idea what went wrong. */
  abort();
}

static apr_status_t
error_fetch(serf_request_t *request,
            report_fetch_t *fetch_ctx,
            svn_error_t *err)
{
  fetch_ctx->err = err;

  fetch_ctx->done = TRUE;

  fetch_ctx->done_item.data = fetch_ctx;
  fetch_ctx->done_item.next = *fetch_ctx->done_list;
  *fetch_ctx->done_list = &fetch_ctx->done_item;

  serf_request_set_handler(request, svn_ra_serf__handler_discard_body, NULL);

  return APR_SUCCESS;
}

static apr_status_t
handle_fetch(serf_request_t *request,
             serf_bucket_t *response,
             void *handler_baton,
             apr_pool_t *pool)
{
  const char *data;
  apr_size_t len;
  serf_status_line sl;
  apr_status_t status;
  report_fetch_t *fetch_ctx = handler_baton;
  svn_error_t *err;

  if (fetch_ctx->read_headers == FALSE)
    {
      serf_bucket_t *hdrs;
      const char *val;
      report_info_t *info;

      hdrs = serf_bucket_response_get_headers(response);
      val = serf_bucket_headers_get(hdrs, "Content-Type");
      info = fetch_ctx->info;

      if (!info->dir->dir_baton)
        {
          err = open_dir(info->dir);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }
        }

      apr_pool_create(&info->editor_pool, info->dir->dir_baton_pool);

      /* Expand our full name now if we haven't done so yet. */
      if (!info->name)
        {
          info->name_buf = svn_stringbuf_dup(info->dir->name_buf,
                                             info->dir->dir_baton_pool);
          svn_path_add_component(info->name_buf, info->base_name);
          info->name = info->name_buf->data;
        }

      if (SVN_IS_VALID_REVNUM(info->base_rev))
        {
          err = info->dir->update_editor->open_file(info->name,
                                          info->dir->dir_baton,
                                          info->base_rev,
                                          info->editor_pool,
                                          &info->file_baton);
        }
      else
        {
          err = info->dir->update_editor->add_file(info->name,
                                          info->dir->dir_baton,
                                          NULL,
                                          info->base_rev,
                                          info->editor_pool,
                                          &info->file_baton);
        }

      if (err)
        {
          return error_fetch(request, fetch_ctx, err);
        }

      err = info->dir->update_editor->apply_textdelta(info->file_baton,
                                                NULL,
                                                info->editor_pool,
                                                &info->textdelta,
                                                &info->textdelta_baton);

      if (err)
        {
          return error_fetch(request, fetch_ctx, err);
        }

      if (val && strcasecmp(val, "application/vnd.svn-svndiff") == 0)
        {
          fetch_ctx->delta_stream =
              svn_txdelta_parse_svndiff(info->textdelta,
                                        info->textdelta_baton,
                                        TRUE, info->editor_pool);
        }
      else
        {
          fetch_ctx->delta_stream = NULL;
        }

      fetch_ctx->read_headers = TRUE;
    }

  while (1)
    {
      svn_txdelta_window_t delta_window = { 0 };
      svn_txdelta_op_t delta_op;
      svn_string_t window_data;

      status = serf_bucket_read(response, 8000, &data, &len);
      if (SERF_BUCKET_READ_ERROR(status))
        {
          return status;
        }

      fetch_ctx->read_size += len;

      if (fetch_ctx->aborted_read == TRUE)
        {
          /* We haven't caught up to where we were before. */
          if (fetch_ctx->read_size < fetch_ctx->aborted_read_size)
            {
              /* Eek.  What did the file shrink or something? */
              if (APR_STATUS_IS_EOF(status))
                {
                  abort();
                }

              /* Skip on to the next iteration of this loop. */
              if (APR_STATUS_IS_EAGAIN(status))
                {
                  return status;
                }
              continue;
            }

          /* Woo-hoo.  We're back. */ 
          fetch_ctx->aborted_read = FALSE;

          /* Increment data and len by the difference. */
          data += fetch_ctx->read_size - fetch_ctx->aborted_read_size;
          len = fetch_ctx->read_size - fetch_ctx->aborted_read_size;
        }
      
      if (fetch_ctx->delta_stream)
        {
          err = svn_stream_write(fetch_ctx->delta_stream, data, &len);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }
        }
      /* otherwise, manually construct the text delta window. */
      else if (len)
        {
          window_data.data = data;
          window_data.len = len;

          delta_op.action_code = svn_txdelta_new;
          delta_op.offset = 0;
          delta_op.length = len;

          delta_window.tview_len = len;
          delta_window.num_ops = 1;
          delta_window.ops = &delta_op;
          delta_window.new_data = &window_data;

          /* write to the file located in the info. */
          err = fetch_ctx->info->textdelta(&delta_window,
                                           fetch_ctx->info->textdelta_baton);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }
        }

      if (APR_STATUS_IS_EOF(status))
        {
          report_info_t *info = fetch_ctx->info;

          err = info->textdelta(NULL, info->textdelta_baton);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }

          /* set all of the properties we received */
          svn_ra_serf__walk_all_props(info->dir->props,
                         info->base_name,
                         info->base_rev,
                         set_file_props,
                         info, info->editor_pool);
          svn_ra_serf__walk_all_props(info->dir->removed_props,
                         info->base_name,
                         info->base_rev,
                         remove_file_props,
                         info, info->editor_pool);
          svn_ra_serf__walk_all_props(info->dir->props,
                         info->url,
                         info->target_rev,
                         set_file_props,
                         info, info->editor_pool);

          err = info->dir->update_editor->close_file(info->file_baton, NULL,
                                                     info->editor_pool);

          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }

          fetch_ctx->done = TRUE;

          fetch_ctx->done_item.data = fetch_ctx;
          fetch_ctx->done_item.next = *fetch_ctx->done_list;
          *fetch_ctx->done_list = &fetch_ctx->done_item;

          /* We're done with our pools. */
          apr_pool_destroy(info->editor_pool);
          apr_pool_destroy(info->pool);

          return status;
        }
      if (APR_STATUS_IS_EAGAIN(status))
        {
          return status;
        }
    }
  /* not reached */
}

static apr_status_t
handle_stream(serf_request_t *request,
              serf_bucket_t *response,
              void *handler_baton,
              apr_pool_t *pool)
{
  report_fetch_t *fetch_ctx = handler_baton;
  serf_status_line sl;

  serf_bucket_response_status(response, &sl);

  /* Woo-hoo.  Nothing here to see.  */
  if (sl.code == 404)
    {
      fetch_ctx->done = TRUE;
      return svn_ra_serf__handler_discard_body(request, response, NULL, pool);
    }

  while (1)
    {
      const char *data;
      apr_size_t len;
      apr_status_t status;

      status = serf_bucket_read(response, 8000, &data, &len);
      if (SERF_BUCKET_READ_ERROR(status))
        {
          return status;
        }

      fetch_ctx->read_size += len;

      if (fetch_ctx->aborted_read == TRUE)
        {
          /* We haven't caught up to where we were before. */
          if (fetch_ctx->read_size < fetch_ctx->aborted_read_size)
            {
              /* Eek.  What did the file shrink or something? */
              if (APR_STATUS_IS_EOF(status))
                {
                  abort();
                }

              /* Skip on to the next iteration of this loop. */
              if (APR_STATUS_IS_EAGAIN(status))
                {
                  return status;
                }
              continue;
            }

          /* Woo-hoo.  We're back. */ 
          fetch_ctx->aborted_read = FALSE;

          /* Increment data and len by the difference. */
          data += fetch_ctx->read_size - fetch_ctx->aborted_read_size;
          len += fetch_ctx->read_size - fetch_ctx->aborted_read_size;
        }
      
      if (len)
        {
          apr_size_t written_len;

          written_len = len;

          svn_stream_write(fetch_ctx->target_stream, data, &written_len);
        }

      if (APR_STATUS_IS_EOF(status))
        {
          fetch_ctx->done = TRUE;
        }

      if (status)
        {
          return status;
        }
    }
  /* not reached */
}

static svn_error_t *
handle_propchange_only(report_info_t *info)
{
  /* Ensure our parent is open. */
  SVN_ERR(open_dir(info->dir));

  apr_pool_create(&info->editor_pool, info->dir->dir_baton_pool);

  /* Expand our full name now if we haven't done so yet. */
  if (!info->name)
    {
      info->name_buf = svn_stringbuf_dup(info->dir->name_buf,
                                         info->dir->dir_baton_pool);
      svn_path_add_component(info->name_buf, info->base_name);
      info->name = info->name_buf->data;
    }

  SVN_ERR(info->dir->update_editor->open_file(info->name,
                                              info->dir->dir_baton,
                                              info->base_rev,
                                              info->editor_pool,
                                              &info->file_baton));

  /* set all of the properties we received */
  svn_ra_serf__walk_all_props(info->dir->props,
                              info->base_name, info->base_rev,
                              set_file_props, info, info->editor_pool);
  svn_ra_serf__walk_all_props(info->dir->removed_props,
                              info->base_name, info->base_rev,
                              remove_file_props, info, info->editor_pool);
  svn_ra_serf__walk_all_props(info->dir->props, info->url, info->target_rev,
                              set_file_props, info, info->editor_pool);

  SVN_ERR(info->dir->update_editor->close_file(info->file_baton, NULL,
                                               info->editor_pool));

  /* We're done with our pools. */
  apr_pool_destroy(info->editor_pool);
  apr_pool_destroy(info->pool);

  info->dir->ref_count--;

  return SVN_NO_ERROR;
}

static void fetch_file(report_context_t *ctx, report_info_t *info)
{
  const char *checked_in_url, *checksum;
  svn_ra_serf__connection_t *conn;<commit_msg>ra_serf: In the case where we're not actually fetching the file, handle the case where we're 'adding' a file or we 'have' text changes (but we're not actually going to deliver them).<commit_after>/*
 * update.c :  entry point for update RA functions for ra_serf
 *
 * ====================================================================
 * Copyright (c) 2006 CollabNet.  All rights reserved.
 *
 * This software is licensed as described in the file COPYING, which
 * you should have received as part of this distribution.  The terms
 * are also available at http://subversion.tigris.org/license-1.html.
 * If newer versions of this license are posted there, you may use a
 * newer version instead, at your option.
 *
 * This software consists of voluntary contributions made by many
 * individuals.  For exact contribution history, see the revision
 * history and logs, available at http://subversion.tigris.org/.
 * ====================================================================
 */




#define APR_WANT_STRFUNC
#include <apr_want.h>

#include <apr_uri.h>

#include <expat.h>

#include <serf.h>

#include "svn_pools.h"
#include "svn_ra.h"
#include "svn_dav.h"
#include "svn_xml.h"
#include "../libsvn_ra/ra_loader.h"
#include "svn_config.h"
#include "svn_delta.h"
#include "svn_version.h"
#include "svn_path.h"
#include "svn_base64.h"
#include "svn_private_config.h"

#include "ra_serf.h"



/*
 * This enum represents the current state of our XML parsing for a REPORT.
 */
typedef enum {
    OPEN_DIR,
    ADD_DIR,
    OPEN_FILE,
    ADD_FILE,
    PROP,
    IGNORE_PROP_NAME,
    NEED_PROP_NAME,
} report_state_e;

/*
 * This structure represents the information for a directory.
 */
typedef struct report_dir_t
{
  /* Our parent directory.
   *
   * This value is NULL when we are the root.
   */
  struct report_dir_t *parent_dir;

  apr_pool_t *pool;

  /* Our name sans any parents. */
  const char *base_name;

  /* the expanded directory name (including all parent names) */
  const char *name;

  /* temporary path buffer for this directory. */
  svn_stringbuf_t *name_buf;

  /* the canonical url for this directory. */
  const char *url;

  /* Our base revision - SVN_INVALID_REVNUM if we're adding this dir. */
  svn_revnum_t base_rev;

  /* The target revision we're retrieving. */
  svn_revnum_t target_rev;

  /* controlling dir baton - this is only created in open_dir() */
  void *dir_baton;
  apr_pool_t *dir_baton_pool;

  /* Our master update editor and baton. */
  const svn_delta_editor_t *update_editor;
  void *update_baton;

  /* How many references to this directory do we still have open? */
  apr_size_t ref_count;

  /* Namespace list allocated out of this ->pool. */
  svn_ra_serf__ns_t *ns_list;

  /* hashtable for all of the properties (shared within a dir) */
  apr_hash_t *props;

  /* hashtable for all to-be-removed properties (shared within a dir) */
  apr_hash_t *removed_props;

  /* The propfind request for our current directory */
  svn_ra_serf__propfind_context_t *propfind;

  /* Has the server told us to fetch the dir props? */
  svn_boolean_t fetch_props;

  /* Have we closed the directory tag (meaning no more additions)? */
  svn_boolean_t tag_closed;

  /* The children of this directory  */
  struct report_dir_t *children;

  /* The next sibling of this directory */
  struct report_dir_t *sibling;
} report_dir_t;

/*
 * This structure represents the information for a file.
 *
 * A directory may have a report_info_t associated with it as well.
 *
 * This structure is created as we parse the REPORT response and
 * once the element is completed, we create a report_fetch_t structure
 * to give to serf to retrieve this file.
 */
typedef struct report_info_t
{
  apr_pool_t *pool;

  /* The enclosing directory.
   *
   * If this structure refers to a directory, the dir it points to will be
   * itself.
   */
  report_dir_t *dir;

  /* Our name sans any directory info. */
  const char *base_name;

  /* the expanded file name (including all parent directory names) */
  const char *name;

  /* file name buffer */
  svn_stringbuf_t *name_buf;

  /* the canonical url for this file. */
  const char *url;

  /* Our base revision - SVN_INVALID_REVNUM if we're adding this file. */
  svn_revnum_t base_rev;

  /* The target revision we're retrieving. */
  svn_revnum_t target_rev;

  /* our delta base, if present (NULL if we're adding the file) */
  const svn_string_t *delta_base;

  /* The propfind request for our current file (if present) */
  svn_ra_serf__propfind_context_t *propfind;

  /* Has the server told us to fetch the file props? */
  svn_boolean_t fetch_props;

  /* Has the server told us to go fetch - only valid if we had it already */
  svn_boolean_t fetch_file;

  /* pool passed to update->add_file, etc. */
  apr_pool_t *editor_pool;

  /* controlling file_baton and textdelta handler */
  void *file_baton;
  svn_txdelta_window_handler_t textdelta;
  void *textdelta_baton;

  /* temporary property for this file which is currently being parsed
   * It will eventually be stored in our parent directory's property hash.
   */
  const char *prop_ns;
  const char *prop_name;
  const char *prop_val;
  apr_size_t prop_val_len;
  const char *prop_encoding;
} report_info_t;

/*
 * This structure represents a single request to GET (fetch) a file with
 * its associated Serf session/connection.
 */
typedef struct report_fetch_t {
  /* Our pool. */
  apr_pool_t *pool;

  /* Non-NULL if we received an error during processing. */
  svn_error_t *err;

  /* The session we should use to fetch the file. */
  svn_ra_serf__session_t *sess;

  /* The connection we should use to fetch file. */
  svn_ra_serf__connection_t *conn;

  /* Stores the information for the file we want to fetch. */
  report_info_t *info;

  /* Have we read our response headers yet? */
  svn_boolean_t read_headers;

  /* This flag is set when our response is aborted before we reach the
   * end and we decide to requeue this request.
   */
  svn_boolean_t aborted_read;
  apr_off_t aborted_read_size;

  /* This is the amount of data that we have read so far. */
  apr_off_t read_size;

  /* If we're receiving an svndiff, this will be non-NULL. */
  svn_stream_t *delta_stream;

  /* If we're writing this file to a stream, this will be non-NULL. */
  svn_stream_t *target_stream;

  /* Are we done fetching this file? */
  svn_boolean_t done;
  svn_ra_serf__list_t **done_list;
  svn_ra_serf__list_t done_item;

} report_fetch_t;

/*
 * Encapsulates all of the REPORT parsing state that we need to know at
 * any given time.
 *
 * Previous states are stored in ->prev field.
 */
typedef struct report_state_list_t {
   /* The current state that we are in now. */
  report_state_e state;

  /* Information */
  report_info_t *info;

  /* Temporary pool */
  apr_pool_t *pool;

  /* Temporary namespace list allocated from ->pool */
  svn_ra_serf__ns_t *ns_list;

  /* The previous state we were in. */
  struct report_state_list_t *prev;
} report_state_list_t;

/*
 * The master structure for a REPORT request and response.
 */
typedef struct {
  apr_pool_t *pool;

  svn_ra_serf__session_t *sess;
  svn_ra_serf__connection_t *conn;

  /* Source path and destination path */
  const char *source;
  const char *destination;

  /* Our update target. */
  const char *update_target;

  /* What is the target revision that we want for this REPORT? */
  svn_revnum_t target_rev;

  /* Have we been asked to ignore ancestry, recursion, or textdeltas? */
  svn_boolean_t ignore_ancestry;
  svn_boolean_t recurse;
  svn_boolean_t text_deltas;

  /* Our master update editor and baton. */
  const svn_delta_editor_t *update_editor;
  void *update_baton;

  /* The request body for the REPORT. */
  serf_bucket_t *buckets;

  /* Our XML parser and root namespace for parsing the response. */
  XML_Parser xmlp;
  svn_ra_serf__ns_t *ns_list;

  /* the current state we are in for parsing the REPORT response.
   *
   * could allocate this as an array rather than a linked list.
   *
   * (We tend to use only about 8 or 9 states in a given update-report,
   * but in theory it could be much larger based on the number of directories
   * we are adding.)
   */
  report_state_list_t *state;
  /* A list of previous states that we have created but aren't using now. */
  report_state_list_t *free_state;

  /* root directory object */
  report_dir_t *root_dir;

  /* number of pending GET requests */
  unsigned int active_fetches;

  /* completed fetches (contains report_fetch_t) */
  svn_ra_serf__list_t *done_fetches;

  /* number of pending PROPFIND requests */
  unsigned int active_propfinds;

  /* completed PROPFIND requests (contains propfind_context_t) */
  svn_ra_serf__list_t *done_propfinds;

  /* list of files that will only have prop changes (contains report_info_t) */
  svn_ra_serf__list_t *file_propchanges_only;

  /* free list of info structures */
  report_info_t *free_info;

  /* The path to the REPORT request */
  const char *path;

  /* Are we done parsing the REPORT response? */
  svn_boolean_t done;

} report_context_t;



static void push_state(report_context_t *ctx, report_state_e state)
{
  report_state_list_t *new_state;

  if (!ctx->free_state)
    {
      new_state = apr_palloc(ctx->sess->pool, sizeof(*ctx->state));

      apr_pool_create(&new_state->pool, ctx->sess->pool);
    }
  else
    {
      new_state = ctx->free_state;
      ctx->free_state = ctx->free_state->prev;

      apr_pool_clear(new_state->pool);
    }
  new_state->state = state;

  if (!ctx->state && state == OPEN_DIR)
    {
      new_state->info = apr_palloc(ctx->sess->pool, sizeof(*new_state->info));
      apr_pool_create(&new_state->info->pool, ctx->sess->pool);

      /* Create our root state now. */
      new_state->info->dir = apr_pcalloc(ctx->sess->pool,
                                        sizeof(*new_state->info->dir));
      new_state->info->dir->pool = new_state->info->pool;

      /* Create the root property tree. */
      new_state->info->dir->props = apr_hash_make(new_state->info->pool);
      new_state->info->dir->removed_props =
          apr_hash_make(new_state->info->pool);

      /* Point to the update_editor */
      new_state->info->dir->update_editor = ctx->update_editor;
      new_state->info->dir->update_baton = ctx->update_baton;

      /* Allow us to be found later. */
      ctx->root_dir = new_state->info->dir;
    }
  else if (state == ADD_DIR || state == OPEN_DIR)
    {
      new_state->info = apr_palloc(ctx->state->info->pool,
                                   sizeof(*new_state->info));
      apr_pool_create(&new_state->info->pool, ctx->state->info->pool);

      new_state->info->dir =
          apr_pcalloc(ctx->state->info->pool, sizeof(*new_state->info->dir));
      new_state->info->dir->pool = new_state->info->pool;
      new_state->info->dir->parent_dir = ctx->state->info->dir;
      new_state->info->dir->parent_dir->ref_count++;

      new_state->info->dir->props = apr_hash_make(new_state->info->pool);
      new_state->info->dir->removed_props =
          apr_hash_make(new_state->info->pool);

      /* Point our ns_list at our parents to try to reuse it. */
      new_state->info->dir->ns_list =
          new_state->info->dir->parent_dir->ns_list;

      /* Point to the update_editor */
      new_state->info->dir->update_editor = ctx->update_editor;
      new_state->info->dir->update_baton = ctx->update_baton;

      /* Add ourselves to our parent's list */
      new_state->info->dir->sibling = ctx->state->info->dir->children;
      ctx->state->info->dir->children = new_state->info->dir;
    }
  else if (state == OPEN_FILE || state == ADD_FILE)
    {
      new_state->info = apr_palloc(ctx->state->info->pool,
                                   sizeof(*new_state->info));
      apr_pool_create(&new_state->info->pool, ctx->state->info->pool);
      new_state->info->file_baton = NULL;

      /* Point at our parent's directory state. */
      new_state->info->dir = ctx->state->info->dir;
      new_state->info->dir->ref_count++;
    }
  /* if we have state info from our parent, reuse it. */
  else if (ctx->state && ctx->state->info)
    {
      new_state->info = ctx->state->info;
    }
  else
    {
      abort();
    }

  if (!ctx->state)
    {
      /* Attach to the root state. */
      new_state->ns_list = ctx->ns_list;
    }
  else
    {
      new_state->ns_list = ctx->state->ns_list;
    }

  /* Add it to the state chain. */
  new_state->prev = ctx->state;
  ctx->state = new_state;
}

static void pop_state(report_context_t *ctx)
{
  report_state_list_t *free_state;
  free_state = ctx->state;
  /* advance the current state */
  ctx->state = ctx->state->prev;
  free_state->prev = ctx->free_state;
  ctx->free_state = free_state;
  ctx->free_state->info = NULL;
}

static svn_error_t *
set_file_props(void *baton,
               const char *ns, apr_ssize_t ns_len,
               const char *name, apr_ssize_t name_len,
               const svn_string_t *val,
               apr_pool_t *pool)
{
  report_info_t *info = baton;
  const svn_delta_editor_t *editor = info->dir->update_editor;

  return svn_ra_serf__set_baton_props(editor->change_file_prop,
                                      info->file_baton,
                                      ns, ns_len, name, name_len, val, pool);
}

static svn_error_t *
set_dir_props(void *baton,
              const char *ns, apr_ssize_t ns_len,
              const char *name, apr_ssize_t name_len,
              const svn_string_t *val,
              apr_pool_t *pool)
{
  report_dir_t *dir = baton;
  return svn_ra_serf__set_baton_props(dir->update_editor->change_dir_prop,
                                      dir->dir_baton,
                                      ns, ns_len, name, name_len, val, pool);
}

static svn_error_t *
remove_file_props(void *baton,
                  const char *ns, apr_ssize_t ns_len,
                  const char *name, apr_ssize_t name_len,
                  const svn_string_t *val,
                  apr_pool_t *pool)
{
  report_info_t *info = baton;
  const svn_delta_editor_t *editor = info->dir->update_editor;

  return svn_ra_serf__set_baton_props(editor->change_file_prop,
                                      info->file_baton,
                                      ns, ns_len, name, name_len, NULL, pool);
}

static svn_error_t *
remove_dir_props(void *baton,
                 const char *ns, apr_ssize_t ns_len,
                 const char *name, apr_ssize_t name_len,
                 const svn_string_t *val,
                 apr_pool_t *pool)
{
  report_dir_t *dir = baton;
  svn_ra_serf__set_baton_props(dir->update_editor->change_dir_prop,
                               dir->dir_baton,
                               ns, ns_len, name, name_len, NULL, pool);
}

static svn_error_t*
open_dir(report_dir_t *dir)
{
  /* if we're already open, return now */
  if (dir->dir_baton)
    {
      return SVN_NO_ERROR;
    }

  if (dir->base_name[0] == '\0')
    {
      apr_pool_create(&dir->dir_baton_pool, dir->pool);

      dir->name_buf = svn_stringbuf_create("", dir->dir_baton_pool);
      dir->name = dir->name_buf->data;

      SVN_ERR(dir->update_editor->open_root(dir->update_baton, dir->base_rev,
                                            dir->dir_baton_pool,
                                            &dir->dir_baton));
    }
  else
    {
      SVN_ERR(open_dir(dir->parent_dir));

      apr_pool_create(&dir->dir_baton_pool, dir->parent_dir->dir_baton_pool);

      /* Expand our name. */
      dir->name_buf = svn_stringbuf_dup(dir->parent_dir->name_buf,
                                        dir->dir_baton_pool);
      svn_path_add_component(dir->name_buf, dir->base_name);
      dir->name = dir->name_buf->data;

      if (SVN_IS_VALID_REVNUM(dir->base_rev))
        {
          SVN_ERR(dir->update_editor->open_directory(dir->name,
                                         dir->parent_dir->dir_baton,
                                         dir->base_rev, dir->dir_baton_pool,
                                         &dir->dir_baton));
        }
      else
        {
          SVN_ERR(dir->update_editor->add_directory(dir->name,
                                                    dir->parent_dir->dir_baton,
                                                    NULL, SVN_INVALID_REVNUM,
                                                    dir->dir_baton_pool,
                                                    &dir->dir_baton));
        }
    }

  return SVN_NO_ERROR;
}

static svn_error_t *
close_dir(report_dir_t *dir)
{
  report_dir_t *prev, *sibling;

  if (dir->ref_count)
    {
      abort();
    }

  svn_ra_serf__walk_all_props(dir->props, dir->base_name, dir->base_rev,
                              set_dir_props, dir, dir->dir_baton_pool);

  svn_ra_serf__walk_all_props(dir->removed_props, dir->base_name,
                              dir->base_rev, remove_dir_props, dir,
                              dir->dir_baton_pool);

  if (dir->fetch_props)
    {
      svn_ra_serf__walk_all_props(dir->props, dir->url, dir->target_rev,
                                  set_dir_props, dir, dir->dir_baton_pool);
    }

  SVN_ERR(dir->update_editor->close_directory(dir->dir_baton,
                                              dir->dir_baton_pool));

  /* remove us from our parent's children list */
  if (dir->parent_dir)
    {
      prev = NULL;
      sibling = dir->parent_dir->children;

      while (sibling != dir)
        {
          prev = sibling;
          sibling = sibling->sibling;
          if (!sibling)
            abort();
        }

      if (!prev)
        {
          dir->parent_dir->children = dir->sibling;
        }
      else
        {
          prev->sibling = dir->sibling;
        }
    }

  apr_pool_destroy(dir->dir_baton_pool);
  apr_pool_destroy(dir->pool);

  return SVN_NO_ERROR;
}

static svn_error_t *close_all_dirs(report_dir_t *dir)
{
  while (dir->children)
    {
      SVN_ERR(close_all_dirs(dir->children));
      dir->ref_count--;
    }

  if (dir->ref_count)
    {
      abort();
    }

  if (!dir->dir_baton)
    {
      SVN_ERR(open_dir(dir));
    }

  return close_dir(dir);
}

static apr_status_t
headers_fetch(serf_bucket_t *headers,
              void *baton,
              apr_pool_t *pool)
{
  report_fetch_t *fetch_ctx = baton;

  /* note that we have old VC URL */
  if (SVN_IS_VALID_REVNUM(fetch_ctx->info->base_rev))
    {
      serf_bucket_headers_setn(headers, SVN_DAV_DELTA_BASE_HEADER,
                               fetch_ctx->info->delta_base->data);
      serf_bucket_headers_setn(headers, "Accept-Encoding",
                               "svndiff1;q=0.9,svndiff;q=0.8");
    }
  else
    {
      serf_bucket_headers_setn(headers, "Accept-Encoding", "gzip");
    }

  return APR_SUCCESS;
}

static apr_status_t
cancel_fetch(serf_request_t *request,
             serf_bucket_t *response,
             int status_code,
             void *baton)
{
  report_fetch_t *fetch_ctx = baton;

  /* Uh-oh.  Our connection died on us.
   *
   * The core ra_serf layer will requeue our request - we just need to note
   * that we got cut off in the middle of our song.
   */
  if (!response)
    {
      /* If we already started the fetch and opened the file handle, we need
       * to hold subsequent read() ops until we get back to where we were
       * before the close and we can then resume the textdelta() calls.
       */
      if (fetch_ctx->read_headers == TRUE)
        {
          if (fetch_ctx->aborted_read == FALSE && fetch_ctx->read_size)
            {
              fetch_ctx->aborted_read = TRUE;
              fetch_ctx->aborted_read_size = fetch_ctx->read_size;
            }
          fetch_ctx->read_size = 0;
        }

      return APR_SUCCESS;
    }

  /* We have no idea what went wrong. */
  abort();
}

static apr_status_t
error_fetch(serf_request_t *request,
            report_fetch_t *fetch_ctx,
            svn_error_t *err)
{
  fetch_ctx->err = err;

  fetch_ctx->done = TRUE;

  fetch_ctx->done_item.data = fetch_ctx;
  fetch_ctx->done_item.next = *fetch_ctx->done_list;
  *fetch_ctx->done_list = &fetch_ctx->done_item;

  serf_request_set_handler(request, svn_ra_serf__handler_discard_body, NULL);

  return APR_SUCCESS;
}

static apr_status_t
handle_fetch(serf_request_t *request,
             serf_bucket_t *response,
             void *handler_baton,
             apr_pool_t *pool)
{
  const char *data;
  apr_size_t len;
  serf_status_line sl;
  apr_status_t status;
  report_fetch_t *fetch_ctx = handler_baton;
  svn_error_t *err;

  if (fetch_ctx->read_headers == FALSE)
    {
      serf_bucket_t *hdrs;
      const char *val;
      report_info_t *info;

      hdrs = serf_bucket_response_get_headers(response);
      val = serf_bucket_headers_get(hdrs, "Content-Type");
      info = fetch_ctx->info;

      if (!info->dir->dir_baton)
        {
          err = open_dir(info->dir);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }
        }

      apr_pool_create(&info->editor_pool, info->dir->dir_baton_pool);

      /* Expand our full name now if we haven't done so yet. */
      if (!info->name)
        {
          info->name_buf = svn_stringbuf_dup(info->dir->name_buf,
                                             info->dir->dir_baton_pool);
          svn_path_add_component(info->name_buf, info->base_name);
          info->name = info->name_buf->data;
        }

      if (SVN_IS_VALID_REVNUM(info->base_rev))
        {
          err = info->dir->update_editor->open_file(info->name,
                                          info->dir->dir_baton,
                                          info->base_rev,
                                          info->editor_pool,
                                          &info->file_baton);
        }
      else
        {
          err = info->dir->update_editor->add_file(info->name,
                                          info->dir->dir_baton,
                                          NULL,
                                          info->base_rev,
                                          info->editor_pool,
                                          &info->file_baton);
        }

      if (err)
        {
          return error_fetch(request, fetch_ctx, err);
        }

      err = info->dir->update_editor->apply_textdelta(info->file_baton,
                                                NULL,
                                                info->editor_pool,
                                                &info->textdelta,
                                                &info->textdelta_baton);

      if (err)
        {
          return error_fetch(request, fetch_ctx, err);
        }

      if (val && strcasecmp(val, "application/vnd.svn-svndiff") == 0)
        {
          fetch_ctx->delta_stream =
              svn_txdelta_parse_svndiff(info->textdelta,
                                        info->textdelta_baton,
                                        TRUE, info->editor_pool);
        }
      else
        {
          fetch_ctx->delta_stream = NULL;
        }

      fetch_ctx->read_headers = TRUE;
    }

  while (1)
    {
      svn_txdelta_window_t delta_window = { 0 };
      svn_txdelta_op_t delta_op;
      svn_string_t window_data;

      status = serf_bucket_read(response, 8000, &data, &len);
      if (SERF_BUCKET_READ_ERROR(status))
        {
          return status;
        }

      fetch_ctx->read_size += len;

      if (fetch_ctx->aborted_read == TRUE)
        {
          /* We haven't caught up to where we were before. */
          if (fetch_ctx->read_size < fetch_ctx->aborted_read_size)
            {
              /* Eek.  What did the file shrink or something? */
              if (APR_STATUS_IS_EOF(status))
                {
                  abort();
                }

              /* Skip on to the next iteration of this loop. */
              if (APR_STATUS_IS_EAGAIN(status))
                {
                  return status;
                }
              continue;
            }

          /* Woo-hoo.  We're back. */ 
          fetch_ctx->aborted_read = FALSE;

          /* Increment data and len by the difference. */
          data += fetch_ctx->read_size - fetch_ctx->aborted_read_size;
          len = fetch_ctx->read_size - fetch_ctx->aborted_read_size;
        }
      
      if (fetch_ctx->delta_stream)
        {
          err = svn_stream_write(fetch_ctx->delta_stream, data, &len);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }
        }
      /* otherwise, manually construct the text delta window. */
      else if (len)
        {
          window_data.data = data;
          window_data.len = len;

          delta_op.action_code = svn_txdelta_new;
          delta_op.offset = 0;
          delta_op.length = len;

          delta_window.tview_len = len;
          delta_window.num_ops = 1;
          delta_window.ops = &delta_op;
          delta_window.new_data = &window_data;

          /* write to the file located in the info. */
          err = fetch_ctx->info->textdelta(&delta_window,
                                           fetch_ctx->info->textdelta_baton);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }
        }

      if (APR_STATUS_IS_EOF(status))
        {
          report_info_t *info = fetch_ctx->info;

          err = info->textdelta(NULL, info->textdelta_baton);
          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }

          /* set all of the properties we received */
          svn_ra_serf__walk_all_props(info->dir->props,
                         info->base_name,
                         info->base_rev,
                         set_file_props,
                         info, info->editor_pool);
          svn_ra_serf__walk_all_props(info->dir->removed_props,
                         info->base_name,
                         info->base_rev,
                         remove_file_props,
                         info, info->editor_pool);
          svn_ra_serf__walk_all_props(info->dir->props,
                         info->url,
                         info->target_rev,
                         set_file_props,
                         info, info->editor_pool);

          err = info->dir->update_editor->close_file(info->file_baton, NULL,
                                                     info->editor_pool);

          if (err)
            {
              return error_fetch(request, fetch_ctx, err);
            }

          fetch_ctx->done = TRUE;

          fetch_ctx->done_item.data = fetch_ctx;
          fetch_ctx->done_item.next = *fetch_ctx->done_list;
          *fetch_ctx->done_list = &fetch_ctx->done_item;

          /* We're done with our pools. */
          apr_pool_destroy(info->editor_pool);
          apr_pool_destroy(info->pool);

          return status;
        }
      if (APR_STATUS_IS_EAGAIN(status))
        {
          return status;
        }
    }
  /* not reached */
}

static apr_status_t
handle_stream(serf_request_t *request,
              serf_bucket_t *response,
              void *handler_baton,
              apr_pool_t *pool)
{
  report_fetch_t *fetch_ctx = handler_baton;
  serf_status_line sl;

  serf_bucket_response_status(response, &sl);

  /* Woo-hoo.  Nothing here to see.  */
  if (sl.code == 404)
    {
      fetch_ctx->done = TRUE;
      return svn_ra_serf__handler_discard_body(request, response, NULL, pool);
    }

  while (1)
    {
      const char *data;
      apr_size_t len;
      apr_status_t status;

      status = serf_bucket_read(response, 8000, &data, &len);
      if (SERF_BUCKET_READ_ERROR(status))
        {
          return status;
        }

      fetch_ctx->read_size += len;

      if (fetch_ctx->aborted_read == TRUE)
        {
          /* We haven't caught up to where we were before. */
          if (fetch_ctx->read_size < fetch_ctx->aborted_read_size)
            {
              /* Eek.  What did the file shrink or something? */
              if (APR_STATUS_IS_EOF(status))
                {
                  abort();
                }

              /* Skip on to the next iteration of this loop. */
              if (APR_STATUS_IS_EAGAIN(status))
                {
                  return status;
                }
              continue;
            }

          /* Woo-hoo.  We're back. */ 
          fetch_ctx->aborted_read = FALSE;

          /* Increment data and len by the difference. */
          data += fetch_ctx->read_size - fetch_ctx->aborted_read_size;
          len += fetch_ctx->read_size - fetch_ctx->aborted_read_size;
        }
      
      if (len)
        {
          apr_size_t written_len;

          written_len = len;

          svn_stream_write(fetch_ctx->target_stream, data, &written_len);
        }

      if (APR_STATUS_IS_EOF(status))
        {
          fetch_ctx->done = TRUE;
        }

      if (status)
        {
          return status;
        }
    }
  /* not reached */
}

static svn_error_t *
handle_propchange_only(report_info_t *info)
{
  /* Ensure our parent is open. */
  SVN_ERR(open_dir(info->dir));

  apr_pool_create(&info->editor_pool, info->dir->dir_baton_pool);

  /* Expand our full name now if we haven't done so yet. */
  if (!info->name)
    {
      info->name_buf = svn_stringbuf_dup(info->dir->name_buf,
                                         info->dir->dir_baton_pool);
      svn_path_add_component(info->name_buf, info->base_name);
      info->name = info->name_buf->data;
    }

  if (SVN_IS_VALID_REVNUM(info->base_rev))
    {
      SVN_ERR(info->dir->update_editor->open_file(info->name,
                                                  info->dir->dir_baton,
                                                  info->base_rev,
                                                  info->editor_pool,
                                                  &info->file_baton));
    }
  else
    {
      SVN_ERR(info->dir->update_editor->add_file(info->name,
                                                 info->dir->dir_baton,
                                                 NULL,
                                                 info->base_rev,
                                                 info->editor_pool,
                                                 &info->file_baton));
    }

  if (info->fetch_file)
    {
      SVN_ERR(info->dir->update_editor->apply_textdelta(info->file_baton,
                                                    NULL,
                                                    info->editor_pool,
                                                    &info->textdelta,
                                                    &info->textdelta_baton));
    }

  /* set all of the properties we received */
  svn_ra_serf__walk_all_props(info->dir->props,
                              info->base_name, info->base_rev,
                              set_file_props, info, info->editor_pool);
  svn_ra_serf__walk_all_props(info->dir->removed_props,
                              info->base_name, info->base_rev,
                              remove_file_props, info, info->editor_pool);
  svn_ra_serf__walk_all_props(info->dir->props, info->url, info->target_rev,
                              set_file_props, info, info->editor_pool);

  SVN_ERR(info->dir->update_editor->close_file(info->file_baton, NULL,
                                               info->editor_pool));

  /* We're done with our pools. */
  apr_pool_destroy(info->editor_pool);
  apr_pool_destroy(info->pool);

  info->dir->ref_count--;

  return SVN_NO_ERROR;
}

static void fetch_file(report_context_t *ctx, report_info_t *info)
{
  const char *checked_in_url, *checksum;
  svn_ra_serf__connection_t *conn;
{'fraction_non_alphanumeric': 0.10560877257295867, 'fraction_numerical': 0.001083820787045154, 'mean_word_length': 1.955485420886523, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 909, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 180, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5545068', 'n_tokens_mistral': 19946, 'n_tokens_neox': 19482, 'n_words': 4466}