// file      : web/apache/stream -*- C++ -*-
// copyright : Copyright (c) 2014-2015 Code Synthesis Ltd
// license   : MIT; see accompanying LICENSE file

#ifndef WEB_APACHE_STREAM
#define WEB_APACHE_STREAM

#include <httpd.h>
#include <http_protocol.h>

#include <ios>       // streamsize
#include <vector>
#include <cstring>   // memmove()
#include <streambuf>
#include <algorithm> // min(), max()

#include <web/module>

namespace web
{
  namespace apache
  {
    // Object of a class implementing this interface is intended for
    // keeping the state of writing response to the client.
    //
    struct write_state
    {
      // Called by ostreambuf methods when some content to be written to the
      // client.
      //
      virtual void
      set_write_state () = 0;

      // Called to check if any data have already been written to the client.
      // Such checks required for some operations which are impossible to
      // execute after response is partially written.
      //
      virtual bool
      get_write_state () const noexcept = 0;
    };

    class ostreambuf: public std::streambuf
    {
    public:
      ostreambuf (request_rec* rec, write_state& ws): rec_ (rec), ws_ (ws) {}

    private:
      virtual int_type
      overflow (int_type c)
      {
        if (c != traits_type::eof ())
        {
          ws_.set_write_state ();

          char chr (c);

          // Throwing allows to distinguish comm failure from other IO error
          // conditions.
          //
          if (ap_rwrite (&chr, sizeof (chr), rec_) == -1)
            throw invalid_request (HTTP_REQUEST_TIME_OUT);
        }

        return c;
      }

      virtual std::streamsize
      xsputn (const char* s, std::streamsize num)
      {
        ws_.set_write_state ();

        if (ap_rwrite (s, num, rec_) < 0)
        {
          throw invalid_request (HTTP_REQUEST_TIME_OUT);
        }

        return num;
      }

      virtual int
      sync ()
      {
        if (ap_rflush (rec_) < 0)
        {
          throw invalid_request (HTTP_REQUEST_TIME_OUT);
        }

        return 0;
      }

    private:
      request_rec* rec_;
      write_state& ws_;
    };

    class istreambuf: public std::streambuf
    {
    public:
      istreambuf (request_rec* rec,
                  write_state& ws,
                  size_t bufsize = 1024,
                  size_t putback = 1)
          : rec_ (rec),
            ws_ (ws),
            bufsize_ (std::max (bufsize, (size_t)1)),
            putback_ (std::min (putback, bufsize_ - 1)),
            buf_ (bufsize_)
      {
        if (ws_.get_write_state ())
        {
          throw sequence_error ("::web::apache::istreambuf::istreambuf");
        }

        char* p (buf_.data () + putback_);
        setg (p, p, p);

        int status (ap_setup_client_block (rec_, REQUEST_CHUNKED_DECHUNK));

        if (status != OK)
        {
          throw invalid_request (status);
        }
      }

    private:
      virtual int_type
      underflow ()
      {
        if (ws_.get_write_state ())
        {
          throw sequence_error ("::web::apache::istreambuf::underflow");
        }

        if (gptr () < egptr ())
          return traits_type::to_int_type (*gptr ());

        size_t pb (std::min ((size_t)(gptr () - eback ()), putback_));
        std::memmove (buf_.data () + putback_ - pb, gptr () - pb, pb);

        char* p (buf_.data () + putback_);
        int rb (ap_get_client_block (rec_, p, bufsize_ - putback_));

        if (rb == 0)
        {
          return traits_type::eof ();
        }

        if (rb < 0)
        {
          throw invalid_request (HTTP_REQUEST_TIME_OUT);
        }

        setg (p - pb, p, p + rb);
        return traits_type::to_int_type (*gptr ());
      }

    private:
      request_rec* rec_;
      write_state& ws_;
      size_t bufsize_;
      size_t putback_;
      std::vector<char> buf_;
    };
  }
}

#endif // WEB_APACHE_STREAM