/***************************************************************************

 *                                  _   _ ____  _

 *  Project                     ___| | | |  _ \| |

 *                             / __| | | | |_) | |

 *                            | (__| |_| |  _ <| |___

 *                             \___|\___/|_| \_\_____|

 *

 * Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.

 *

 * 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 https://curl.se/docs/copyright.html.

 *

 * You may opt to use, copy, modify, merge, publish, distribute and/or sell

 * copies of the Software, and permit persons to whom the Software is

 * furnished to do so, under the terms of the COPYING file.

 *

 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY

 * KIND, either express or implied.

 *

 * SPDX-License-Identifier: curl

 *

 ***************************************************************************/

#include "curl_setup.h"

#include "urldata.h"

#include "cfilters.h"

#include "multiif.h"

#include "sendf.h"

#include "curl_trc.h"

#include "transfer.h"

#include "cw-out.h"

#include "cw-pause.h"

/**

 * OVERALL DESIGN of this client writer

 *

 * The 'cw-out' writer is supposed to be the last writer in a transfer's

 * stack. It is always added when that stack is initialized. Its purpose

 * is to pass BODY and HEADER bytes to the client-installed callback

 * functions.

 *

 * These callback may return `CURL_WRITEFUNC_PAUSE` to indicate that the

 * data had not been written and the whole transfer should stop receiving

 * new data. Or at least, stop calling the functions. When the transfer

 * is "unpaused" by the client, the previous data shall be passed as

 * if nothing happened.

 *

 * The `cw-out` writer therefore manages buffers for bytes that could

 * not be written. Data that was already in flight from the server also

 * needs buffering on paused transfer when it arrives.

 *

 * In addition, the writer allows buffering of "small" body writes,

 * so client functions are called less often. That is only enabled on a

 * number of conditions.

 *

 * HEADER and BODY data may arrive in any order. For paused transfers,

 * a list of `struct cw_out_buf` is kept for `cw_out_type` types. The

 * list may be: [BODY]->[HEADER]->[BODY]->[HEADER]....

 * When unpausing, this list is "played back" to the client callbacks.

 *

 * The amount of bytes being buffered is limited by `DYN_PAUSE_BUFFER`

 * and when that is exceeded `CURLE_TOO_LARGE` is returned as error.

 */

typedef enum {

  CW_OUT_NONE,

  CW_OUT_BODY,

  CW_OUT_BODY_0LEN,

  CW_OUT_HDS

} cw_out_type;

struct cw_out_buf {

  struct cw_out_buf *next;

  struct dynbuf b;

  cw_out_type type;

};

static struct cw_out_buf *cw_out_buf_create(cw_out_type otype)

{

  struct cw_out_buf *cwbuf = curlx_calloc(1, sizeof(*cwbuf));

  if(cwbuf) {

    cwbuf->type = otype;

    curlx_dyn_init(&cwbuf->b, DYN_PAUSE_BUFFER);

  }

  return cwbuf;

}

static void cw_out_buf_free(struct cw_out_buf *cwbuf)

{

  if(cwbuf) {

    curlx_dyn_free(&cwbuf->b);

    curlx_free(cwbuf);

  }

}

struct cw_out_ctx {

  struct Curl_cwriter super;

  struct cw_out_buf *buf;

  BIT(paused);

  BIT(errored);

};

static CURLcode cw_out_init(struct Curl_easy *data,

                            struct Curl_cwriter *writer)

{

  struct cw_out_ctx *ctx = writer->ctx;

  (void)data;

  ctx->buf = NULL;

  return CURLE_OK;

}

static void cw_out_bufs_free(struct cw_out_ctx *ctx)

{

  while(ctx->buf) {

    struct cw_out_buf *next = ctx->buf->next;

    cw_out_buf_free(ctx->buf);

    ctx->buf = next;

  }

}

static size_t cw_out_bufs_len(struct cw_out_ctx *ctx)

{

  struct cw_out_buf *cwbuf = ctx->buf;

  size_t len = 0;

  while(cwbuf) {

    len += curlx_dyn_len(&cwbuf->b);

    cwbuf = cwbuf->next;

  }

  return len;

}

static void cw_out_close(struct Curl_easy *data, struct Curl_cwriter *writer)

{

  struct cw_out_ctx *ctx = writer->ctx;

  (void)data;

  cw_out_bufs_free(ctx);

}

/**

 * Return the current curl_write_callback and user_data for the buf type

 */

static void cw_get_writefunc(struct Curl_easy *data, cw_out_type otype,

                             curl_write_callback *pwcb, void **pwcb_data,

                             size_t *pmax_write, size_t *pmin_write)

{

  switch(otype) {

  case CW_OUT_BODY:

  case CW_OUT_BODY_0LEN:

    *pwcb = data->set.fwrite_func;

    *pwcb_data = data->set.out;

    *pmax_write = CURL_MAX_WRITE_SIZE;

    /* if we ever want buffering of BODY output, we can set `min_write`

     * the preferred size. The default should always be to pass data

     * to the client as it comes without delay */

    *pmin_write = 0;

    break;

  case CW_OUT_HDS:

    *pwcb = data->set.fwrite_header ? data->set.fwrite_header :

             (data->set.writeheader ? data->set.fwrite_func : NULL);

    *pwcb_data = data->set.writeheader;

    *pmax_write = 0; /* do not chunk-write headers, write them as they are */

    *pmin_write = 0;

    break;

  default:

    *pwcb = NULL;

    *pwcb_data = NULL;

    *pmax_write = CURL_MAX_WRITE_SIZE;

    *pmin_write = 0;

  }

}

static CURLcode cw_out_cb_write(struct cw_out_ctx *ctx,

                                struct Curl_easy *data,

                                curl_write_callback wcb,

                                void *wcb_data,

                                cw_out_type otype,

                                const char *buf, size_t blen,

                                size_t *pnwritten)

{

  size_t nwritten;

  CURLcode result;

  DEBUGASSERT(data->conn);

  *pnwritten = 0;

  Curl_set_in_callback(data, TRUE);

  nwritten = wcb((char *)CURL_UNCONST(buf), 1, blen, wcb_data);

  Curl_set_in_callback(data, FALSE);

  CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes -> %zu",

                 blen, (otype == CW_OUT_HDS) ? "header" : "body",

                 nwritten);

  if(CURL_WRITEFUNC_PAUSE == nwritten) {

    if(data->conn->handler->flags & PROTOPT_NONETWORK) {

      /* Protocols that work without network cannot be paused. This is

         actually only FILE:// just now, and it cannot pause since the

         transfer is not done using the "normal" procedure. */

      failf(data, "Write callback asked for PAUSE when not supported");

      return CURLE_WRITE_ERROR;

    }

    ctx->paused = TRUE;

    CURL_TRC_WRITE(data, "[OUT] PAUSE requested by client");

    result = Curl_xfer_pause_recv(data, TRUE);

    return result ? result : CURLE_AGAIN;

  }

  else if(CURL_WRITEFUNC_ERROR == nwritten) {

    failf(data, "client returned ERROR on write of %zu bytes", blen);

    return CURLE_WRITE_ERROR;

  }

  else if(nwritten != blen) {

    failf(data, "Failure writing output to destination, "

          "passed %zu returned %zd", blen, nwritten);

    return CURLE_WRITE_ERROR;

  }

  *pnwritten = nwritten;

  return CURLE_OK;

}

static CURLcode cw_out_ptr_flush(struct cw_out_ctx *ctx,

                                 struct Curl_easy *data,

                                 cw_out_type otype,

                                 bool flush_all,

                                 const char *buf, size_t blen,

                                 size_t *pconsumed)

{

  curl_write_callback wcb = NULL;

  void *wcb_data;

  size_t max_write, min_write;

  size_t wlen, nwritten;

  CURLcode result;

  /* If we errored once, we do not invoke the client callback again */

  if(ctx->errored)

    return CURLE_WRITE_ERROR;

  /* write callbacks may get NULLed by the client between calls. */

  cw_get_writefunc(data, otype, &wcb, &wcb_data, &max_write, &min_write);

  if(!wcb) {

    *pconsumed = blen;

    return CURLE_OK;

  }

  *pconsumed = 0;

  if(otype == CW_OUT_BODY_0LEN) {

    DEBUGASSERT(!blen);

    return cw_out_cb_write(ctx, data, wcb, wcb_data, otype,

                           buf, blen, &nwritten);

  }

  else {

    while(blen && !ctx->paused) {

      if(!flush_all && blen < min_write)

        break;

      wlen = max_write ? CURLMIN(blen, max_write) : blen;

      result = cw_out_cb_write(ctx, data, wcb, wcb_data, otype,

                               buf, wlen, &nwritten);

      if(result)

        return result;

      *pconsumed += nwritten;

      blen -= nwritten;

      buf += nwritten;

    }

  }

  return CURLE_OK;

}

static CURLcode cw_out_buf_flush(struct cw_out_ctx *ctx,

                                 struct Curl_easy *data,

                                 struct cw_out_buf *cwbuf,

                                 bool flush_all)

{

  CURLcode result = CURLE_OK;

  if(curlx_dyn_len(&cwbuf->b) || (cwbuf->type == CW_OUT_BODY_0LEN)) {

    size_t consumed;

    result = cw_out_ptr_flush(ctx, data, cwbuf->type, flush_all,

                              curlx_dyn_ptr(&cwbuf->b),

                              curlx_dyn_len(&cwbuf->b),

                              &consumed);

    if(result && (result != CURLE_AGAIN))

      return result;

    result = CURLE_OK;

    if(consumed) {

      if(consumed == curlx_dyn_len(&cwbuf->b)) {

        curlx_dyn_free(&cwbuf->b);

      }

      else {

        DEBUGASSERT(consumed < curlx_dyn_len(&cwbuf->b));

        result = curlx_dyn_tail(&cwbuf->b,

                                curlx_dyn_len(&cwbuf->b) - consumed);

        if(result)

          return result;

      }

    }

  }

  return result;

}

static CURLcode cw_out_flush_chain(struct cw_out_ctx *ctx,

                                   struct Curl_easy *data,

                                   struct cw_out_buf **pcwbuf,

                                   bool flush_all)

{

  struct cw_out_buf *cwbuf = *pcwbuf;

  CURLcode result;

  if(!cwbuf)

    return CURLE_OK;

  if(ctx->paused)

    return CURLE_OK;

  /* write the end of the chain until it blocks or gets empty */

  while(cwbuf->next) {

    struct cw_out_buf **plast = &cwbuf->next;

    while((*plast)->next)

      plast = &(*plast)->next;

    result = cw_out_flush_chain(ctx, data, plast, flush_all);

    if(result)

      return result;

    if(*plast) {

      /* could not write last, paused again? */

      DEBUGASSERT(ctx->paused);

      return CURLE_OK;

    }

  }

  result = cw_out_buf_flush(ctx, data, cwbuf, flush_all);

  if(result)

    return result;

  if(!curlx_dyn_len(&cwbuf->b)) {

    cw_out_buf_free(cwbuf);

    *pcwbuf = NULL;

  }

  return CURLE_OK;

}

static CURLcode cw_out_append(struct cw_out_ctx *ctx,

                              struct Curl_easy *data,

                              cw_out_type otype,

                              const char *buf, size_t blen)

{

  CURL_TRC_WRITE(data, "[OUT] paused, buffering %zu more bytes (%zu/%d)",

                 blen, cw_out_bufs_len(ctx), DYN_PAUSE_BUFFER);

  if(cw_out_bufs_len(ctx) + blen > DYN_PAUSE_BUFFER) {

    failf(data, "pause buffer not large enough -> CURLE_TOO_LARGE");

    return CURLE_TOO_LARGE;

  }

  /* if we do not have a buffer, or it is of another type, make a new one.

   * And for CW_OUT_HDS always make a new one, so we "replay" headers

   * exactly as they came in */

  if(!ctx->buf || (ctx->buf->type != otype) || (otype == CW_OUT_HDS)) {

    struct cw_out_buf *cwbuf = cw_out_buf_create(otype);

    if(!cwbuf)

      return CURLE_OUT_OF_MEMORY;

    cwbuf->next = ctx->buf;

    ctx->buf = cwbuf;

  }

  DEBUGASSERT(ctx->buf && (ctx->buf->type == otype));

  return curlx_dyn_addn(&ctx->buf->b, buf, blen);

}

static CURLcode cw_out_do_write(struct cw_out_ctx *ctx,

                                struct Curl_easy *data,

                                cw_out_type otype,

                                bool flush_all,

                                const char *buf, size_t blen)

{

  CURLcode result = CURLE_OK;

  /* if we have buffered data and it is a different type than what

   * we are writing now, try to flush all */

  if(ctx->buf && ctx->buf->type != otype) {

    result = cw_out_flush_chain(ctx, data, &ctx->buf, TRUE);

    if(result)

      goto out;

  }

  if(ctx->buf) {

    /* still have buffered data, append and flush */

    result = cw_out_append(ctx, data, otype, buf, blen);

    if(result)

      goto out;

    result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all);

    if(result)

      goto out;

  }

  else {

    /* nothing buffered, try direct write */

    size_t consumed;

    result = cw_out_ptr_flush(ctx, data, otype, flush_all,

                              buf, blen, &consumed);

    if(result && (result != CURLE_AGAIN))

      return result;

    result = CURLE_OK;

    if(consumed < blen) {

      /* did not write all, append the rest */

      result = cw_out_append(ctx, data, otype,

                             buf + consumed, blen - consumed);

      if(result)

        goto out;

    }

  }

out:

  if(result) {

    /* We do not want to invoked client callbacks a second time after

     * encountering an error. See issue #13337 */

    ctx->errored = TRUE;

    cw_out_bufs_free(ctx);

  }

  return result;

}

static CURLcode cw_out_write(struct Curl_easy *data,

                             struct Curl_cwriter *writer, int type,

                             const char *buf, size_t blen)

{

  struct cw_out_ctx *ctx = writer->ctx;

  CURLcode result;

  bool flush_all = !!(type & CLIENTWRITE_EOS);

  if((type & CLIENTWRITE_BODY) ||

     ((type & CLIENTWRITE_HEADER) && data->set.include_header)) {

    cw_out_type otype = (!blen && (type & CLIENTWRITE_0LEN)) ?

                        CW_OUT_BODY_0LEN : CW_OUT_BODY;

    result = cw_out_do_write(ctx, data, otype, flush_all, buf, blen);

    if(result)

      return result;

  }

  if(type & (CLIENTWRITE_HEADER | CLIENTWRITE_INFO)) {

    result = cw_out_do_write(ctx, data, CW_OUT_HDS, flush_all, buf, blen);

    if(result)

      return result;

  }

  return CURLE_OK;

}

const struct Curl_cwtype Curl_cwt_out = {

  "cw-out",

  NULL,

  cw_out_init,

  cw_out_write,

  cw_out_close,

  sizeof(struct cw_out_ctx)

};

bool Curl_cw_out_is_paused(struct Curl_easy *data)

{

  struct Curl_cwriter *cw_out;

  struct cw_out_ctx *ctx;

  cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out);

  if(!cw_out)

    return FALSE;

  ctx = (struct cw_out_ctx *)cw_out;

  return ctx->paused;

}

static CURLcode cw_out_flush(struct Curl_easy *data,

                             struct Curl_cwriter *cw_out,

                             bool flush_all)

{

  struct cw_out_ctx *ctx = (struct cw_out_ctx *)cw_out;

  CURLcode result = CURLE_OK;

  if(ctx->errored)

    return CURLE_WRITE_ERROR;

  if(ctx->paused)

    return CURLE_OK;  /* not doing it */

  result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all);

  if(result) {

    ctx->errored = TRUE;

    cw_out_bufs_free(ctx);

    return result;

  }

  return result;

}

CURLcode Curl_cw_out_unpause(struct Curl_easy *data)

{

  struct Curl_cwriter *cw_out;

  CURLcode result = CURLE_OK;

  cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out);

  if(cw_out) {

    struct cw_out_ctx *ctx = (struct cw_out_ctx *)cw_out;

    CURL_TRC_WRITE(data, "[OUT] unpause");

    ctx->paused = FALSE;

    result = Curl_cw_pause_flush(data);

    if(!result)

      result = cw_out_flush(data, cw_out, FALSE);

  }

  return result;

}

CURLcode Curl_cw_out_done(struct Curl_easy *data)

{

  struct Curl_cwriter *cw_out;

  CURLcode result = CURLE_OK;

  cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out);

  if(cw_out) {

    CURL_TRC_WRITE(data, "[OUT] done");

    result = Curl_cw_pause_flush(data);

    if(!result)

      result = cw_out_flush(data, cw_out, TRUE);

  }

  return result;

}