sockbuf.h

Go to the documentation of this file.

/*
  Copyright (c) Mircea Neacsu (2014-2025) Licensed under MIT License.
  This file is part of MLIB project. See LICENSE file for full license terms.
*/

 
#pragma once
 
#if __has_include("defs.h")
#include "defs.h"
#endif
 
#include "safe_winsock.h"
#include "sock.h"
#include "trace.h"
#include <iostream>
 
namespace mlib {
 
#if !defined(SOCKBUF_BUFSIZ)
#define SOCKBUF_BUFSIZ 1024
#endif

 
template <typename T>
class sockbuf : public sock, public std::basic_streambuf<T>
{
public:
  // constructors

  sockbuf () 
    : sock ()
    , x_flags (0)
  {
    setbuf (0, SOCKBUF_BUFSIZ);
  }

  sockbuf (sock::type t, int domain, int proto)
    : sock(t, domain, proto)
    , x_flags (0)
  {
    setbuf (0, SOCKBUF_BUFSIZ);
  }

  sockbuf (const sock& s)
    : sock (s)
    , x_flags (0)
  {
    setbuf (0, SOCKBUF_BUFSIZ);
  }

  sockbuf (const sockbuf<T>& sb)
    : sock (sb)
    , x_flags (sb.x_flags)
  {
    if (x_flags & flags::allocbuf)
      setbuf (0, (int)(const_cast<sockbuf<T>&> (sb).epptr () - const_cast<sockbuf<T>&> (sb).pbase ()));
  }

  sockbuf<T>& operator= (const sockbuf<T>& rhs)
  {
    sock::operator = (rhs);
    std::basic_streambuf<T>::operator= (rhs);
    x_flags = (x_flags & flags::allocbuf) | (rhs.x_flags & ~flags::allocbuf);
    return *this;
  }

  ~sockbuf ()
  {
    if (is_open ())
    {
      overflow ();
      shutdown (shut_readwrite).deactivate (); // ignore any errors
      close ();
    }
 
    if (x_flags & flags::allocbuf)
    {
      delete[] std::basic_streambuf<T>::pbase ();
      delete[] std::basic_streambuf<T>::eback ();
    }
  }
 
   
protected:
  using int_type = std::char_traits<T>::int_type;
 
  int_type underflow () override
  {
    if (x_flags & flags::no_reads)
      return EOF; // reads blocked for this socket
 
    if (std::basic_streambuf<T>::gptr () < std::basic_streambuf<T>::egptr ())
      return *(unsigned char*)std::basic_streambuf<T>::gptr (); // return available char from buffer
 
    if (std::basic_streambuf<T>::eback ())
    {
      // buffered mode
      size_t rval = recv (std::basic_streambuf<T>::eback (), ibsize);
      if (rval != (size_t)EOF)
      {
        std::basic_streambuf<T>::setg (std::basic_streambuf<T>::eback (),
                                       std::basic_streambuf<T>::eback (),
                                       std::basic_streambuf<T>::eback () + rval);
        return rval ? *(unsigned char*)std::basic_streambuf<T>::gptr () : EOF;
      }
      return EOF;
    }
    else
    {
      // unbuffered mode
      unsigned char ch;
      size_t rval = recv (&ch, 1);
      if (rval == (size_t)EOF || rval == 0)
        return EOF;
      return ch;
    }
  }

  int_type overflow (int_type c = EOF) override
  {
    if (x_flags & flags::no_writes)
      return EOF; // socket blocked for writing, can't do anything
 
    if (sync () == EOF) // flush output
      return EOF;
 
    if (c == EOF)
      return EOF;
 
    if (std::basic_streambuf<T>::pbase ())
    {
      // buffered mode
      *std::basic_streambuf<T>::pptr () = (T)c;
      std::basic_streambuf<T>::pbump (1);
    }
    else
    {
      // unbuffered mode
      unsigned char ch = (unsigned char)c;
      size_t rval = send (&ch, 1);
      if (rval == (size_t)EOF || rval == 0)
        return EOF;
    }
    return c;
  }

  int sync () override
  {
    if (std::basic_streambuf<T>::pptr () <= std::basic_streambuf<T>::pbase ())
      return 0; // unbuffered or empty buffer
    if (!(x_flags & flags::no_writes))
    {
      size_t wlen = std::basic_streambuf<T>::pptr () - std::basic_streambuf<T>::pbase ();
      size_t wval = send (std::basic_streambuf<T>::pbase (), wlen);
      std::basic_streambuf<T>::setp (std::basic_streambuf<T>::pbase (),
                                     std::basic_streambuf<T>::epptr ());
      if (wval == wlen)
        return 0;
      TRACE ("sockbuf::sync failed - wanted %d sent %d", wlen, wval);
    }
    return -1;
  }

  std::basic_streambuf<T>* setbuf (T* buf, std::streamsize sz) override
  {
    overflow (EOF); // flush current output buffer
 
    // switch to unbuffered mode
    if (x_flags & flags::allocbuf)
    {
      delete[] std::basic_streambuf<T>::eback ();
      delete[] std::basic_streambuf<T>::pbase ();
    }
    std::basic_streambuf<T>::setp (NULL, NULL);
    std::basic_streambuf<T>::setg (NULL, NULL, NULL);
    x_flags &= ~flags::allocbuf;
    ibsize = 0;
    if (!buf)
    {
      // automatic buffer - allocate separate input and output buffers
      x_flags |= flags::allocbuf;
      T* ptr = new T[(size_t)sz + 1];
      std::basic_streambuf<T>::setp (ptr, ptr + sz);
      ptr = new T[(size_t)sz];
      std::basic_streambuf<T>::setg (ptr, ptr + sz, ptr + sz);
      ibsize = (int)sz;
    }
    else
      std::basic_streambuf<T>::setp (buf,
                                     buf + sz - 1); // user specified buffer used only for output
    return this;
  }

  std::streamsize showmanyc () override
  {
    return nread ();
  }

private:
  enum flags
  {
    allocbuf = 0x0002,  
    no_reads = 0x0004,  
    no_writes = 0x0008, 
    eof_seen = 0x0010   
  };
 
  int x_flags;
  int ibsize; // input buffer size
};
 
} // namespace mlib