StreamBuffer.h

Go to the documentation of this file.

/***********************************************************************************\
* (c) Copyright 1998-2025 CERN for the benefit of the LHCb and ATLAS collaborations *
*                                                                                   *
* This software is distributed under the terms of the Apache version 2 licence,     *
* copied verbatim in the file "LICENSE".                                            *
*                                                                                   *
* In applying this licence, CERN does not waive the privileges and immunities       *
* granted to it by virtue of its status as an Intergovernmental Organization        *
* or submit itself to any jurisdiction.                                             *
\***********************************************************************************/
#pragma once
 
#include <GaudiKernel/Kernel.h>
#include <algorithm>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <list>
#include <string>
#include <typeinfo>
#include <unistd.h>
#include <vector>
 
class StreamBuffer;
class DataObject;
class ContainedObject;

class StreamBuffer /* : public std::string  */
{
public:
  class DataIO {
  public:
    DataIO() = default;
    virtual ~DataIO() = default;
    void badStreamMode() { throw( "Not acceptable stream mode!" ); }
    virtual void serialize( StreamBuffer& stream ) {
      if ( stream.isReading() )
        load( stream );
      else if ( stream.isWriting() )
        dump( stream );
      else
        badStreamMode();
    }

    virtual void load( StreamBuffer& ) { badStreamMode(); }
    virtual void dump( StreamBuffer& ) { badStreamMode(); }
  };

  class Istream : public DataIO {
    std::istream* m_stream;
 
  public:
    Istream( std::istream& str ) : m_stream( &str ) {}

    void load( StreamBuffer& stream ) override {
      // Generic implementation for istreams:
      int len;
      ( *m_stream ) >> len;
      stream.erase();
      stream.reserve( len );
      m_stream->read( stream.data(), len );
    }
  };

  class Ostream : public DataIO {
    std::ostream* m_stream;
 
  public:
    Ostream( std::ostream& str ) : m_stream( &str ) {}
    virtual ~Ostream() = default;

    void dump( StreamBuffer& stream ) override {
      // Generic implementation for ostreams:
      ( *m_stream ) << stream.buffPointer();
      m_stream->write( stream.data(), stream.buffPointer() );
    }
  };
 
public:
  enum Mode { UNINITIALIZED, READING, WRITING };
  enum SwapAction { SINGLE_BYTE, SWAP, NOSWAP };
  enum State { INVALID = -1, VALID };
  class ContainedLink {
  public:
    ContainedObject* first  = nullptr;
    long             second = INVALID;
    long             third  = INVALID;
    ContainedLink()         = default;
    ContainedLink( ContainedObject* pObj, long hint, long link ) : first( pObj ), second( hint ), third( link ) {}
    ContainedLink( const ContainedLink& copy )            = default;
    ContainedLink& operator=( const ContainedLink& copy ) = default;
  };

  class IdentifiedLink {
  public:
    DataObject* first  = nullptr;
    long        second = INVALID;
    IdentifiedLink()   = default;
    IdentifiedLink( DataObject* pObj, long hint ) : first( pObj ), second( hint ) {}
    IdentifiedLink( const IdentifiedLink& copy )            = default;
    IdentifiedLink& operator=( const IdentifiedLink& copy ) = default;
  };
 
  typedef std::vector<ContainedLink> ContainedLinks;
  typedef std::vector<IdentifiedLink> IdentifiedLinks;
  typedef void ( *AnalyzeFunction )( const void* data, int siz, const std::type_info& type );
  friend class DataObject;
 
protected:
  Mode m_mode = UNINITIALIZED;

  mutable long m_pointer = 0;

  mutable long m_length = 0;

  mutable char* m_buffer = nullptr;

  bool m_swapEnabled = true;

  mutable ContainedLinks m_containedLinks;

  mutable IdentifiedLinks m_identifiedLinks;

  AnalyzeFunction m_analyzer = nullptr;

  SwapAction swapBuffer( int siz ) const;

  template <class TYPE>
  StreamBuffer& getObjectPointer( const DataObject* /*pObject*/, TYPE*& refpObject ) {
    IdentifiedLink& link = m_identifiedLinks.back();
    DataObject*     pObj = link.first;
    m_identifiedLinks.pop_back();
    refpObject = dynamic_cast<TYPE*>( pObj );
    return *this;
  }

  template <class TYPE>
  StreamBuffer& getObjectPointer( const ContainedObject* /*pObject*/, TYPE*& refpObject ) {
    ContainedLink&   link = m_containedLinks.back();
    ContainedObject* pObj = link.first;
    m_containedLinks.pop_back();
    refpObject = dynamic_cast<TYPE*>( pObj );
    return *this;
  }
 
public:
  StreamBuffer( bool do_swap = true ) : m_swapEnabled( do_swap ) {}
  virtual ~StreamBuffer() { ::free( m_buffer ); }
  StreamBuffer( const StreamBuffer& ) = delete;
  StreamBuffer& operator=( const StreamBuffer& ) = delete;
  const char* data() const { return m_buffer; }
  char* data() { return m_buffer; }
  void erase() { m_pointer = 0; }
  char* adopt() const {
    char* ptr = m_buffer;
    m_containedLinks.erase( m_containedLinks.begin(), m_containedLinks.end() );
    m_identifiedLinks.erase( m_identifiedLinks.begin(), m_identifiedLinks.end() );
    m_buffer  = NULL; // char *
    m_pointer = 0;    // long
    m_length  = 0;    // long
    return ptr;
  }

  void reserve( long len ) {
    if ( len > m_length ) {
      m_length = ( len < 16384 ) ? 16384 : len;
      m_buffer = (char*)::realloc( m_buffer, m_length );
    }
  }

  void extend( long len ) {
    if ( len + m_pointer > m_length ) {
      // We have to be a bit generous here in order not to run too often
      // into ::realloc().
      long new_len = ( m_length < 16384 ) ? 16384 : 2 * m_length;
      if ( m_length < len ) new_len += len;
      reserve( new_len );
    }
  }

  long size() const { return m_length; }
  ContainedLinks& containedLinks() { return m_containedLinks; }
  const ContainedLinks& containedLinks() const { return m_containedLinks; }

  IdentifiedLinks& identifiedLinks() { return m_identifiedLinks; }
  const IdentifiedLinks& identifiedLinks() const { return m_identifiedLinks; }

  void setMode( Mode m ) {
    m_mode    = m;
    m_pointer = 0;
    m_containedLinks.erase( m_containedLinks.begin(), m_containedLinks.end() );
    m_identifiedLinks.erase( m_identifiedLinks.begin(), m_identifiedLinks.end() );
  }

  bool isReading() const { return m_mode == READING; }

  bool isWriting() const { return m_mode == WRITING; }
  long buffPointer() const { return m_pointer; }
  void setBuffPointer( long ptr ) { m_pointer = ptr; }
  void setAnalyzer( AnalyzeFunction fun = nullptr ) { m_analyzer = fun; }
  void swapToBuffer( const void* source, int siz );

  void swapFromBuffer( void* target, int siz );

  StreamBuffer& writeBytes( const char* str, long len ) {
    extend( m_pointer + len + 4 );
    *this << len;
    std::copy_n( str, len, data() + buffPointer() );
    m_pointer += len;
    return *this;
  }
 
  void getIdentifiedLink( DataObject*& pObject, long& hint ) {
    IdentifiedLink& l = m_identifiedLinks.back();
    pObject           = l.first;
    hint              = l.second;
    m_identifiedLinks.pop_back();
  }
  void addIdentifiedLink( const DataObject* pObject, long hint ) {
    m_identifiedLinks.push_back( IdentifiedLink( const_cast<DataObject*>( pObject ), hint ) );
  }
 
  void getContainedLink( ContainedObject*& pObject, long& hint, long& link ) {
    ContainedLink& l = m_containedLinks.back();
    pObject          = l.first;
    hint             = l.second;
    link             = l.third;
    m_containedLinks.pop_back();
  }
  void addContainedLink( const ContainedObject* pObject, long hint, long link ) {
    m_containedLinks.push_back( ContainedLink( const_cast<ContainedObject*>( pObject ), hint, link ) );
  }
 
#ifdef USE_STREAM_ANALYSER
#  define STREAM_ANALYSE( data, len )                                                                                  \
    if ( 0 != m_analyzer ) m_analyzer( &data, len, typeid( data ) )
#else
#  define STREAM_ANALYSE( data, len )
#endif
 
// Implement streamer macros for primivive data types.
#define IMPLEMENT_STREAMER( TYPE )                                                                                     \
  /*  Output Streamer                                   */                                                             \
  StreamBuffer& operator<<( TYPE data ) {                                                                              \
    swapToBuffer( &data, sizeof( data ) );                                                                             \
    STREAM_ANALYSE( data, sizeof( data ) );                                                                            \
    return *this;                                                                                                      \
  }                                                                                                                    \
  /*  Input Streamer                                    */                                                             \
  StreamBuffer& operator>>( TYPE& data ) {                                                                             \
    swapFromBuffer( &data, sizeof( data ) );                                                                           \
    return *this;                                                                                                      \
  }
// RootCint does not understand this macro....
// But we can easily live without it!
#undef IMPLEMENT_STREAMER

  StreamBuffer& operator<<( long long data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( long long& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( int data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( int& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( unsigned int data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( unsigned int& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( long data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( long& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( unsigned long data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( unsigned long& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( short data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( short& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( unsigned short data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( unsigned short& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( char data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( char& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( unsigned char data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( unsigned char& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( float data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( float& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator<<( double data ) {
    swapToBuffer( &data, sizeof( data ) );
    STREAM_ANALYSE( data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( double& data ) {
    swapFromBuffer( &data, sizeof( data ) );
    return *this;
  }

  StreamBuffer& operator>>( char* data ) {
    long i, len;
    *this >> len;
    for ( i = 0, data[0] = 0; i < len; i++ ) { data[i] = m_buffer[m_pointer++]; }
    return *this;
  }

  StreamBuffer& operator<<( const char* data ) {
    const char* ptr = 0 == data ? "" : data;
    size_t      len = strlen( ptr ) + 1;
    if ( 0 == m_analyzer )
      writeBytes( ptr, len );
    else { STREAM_ANALYSE( data, len ); }
    return *this;
  }

  StreamBuffer& operator>>( std::string& data ) {
    long i, len;
    *this >> len;
    for ( i = 0, data = ""; i < len; i++ ) { data.append( 1, m_buffer[m_pointer++] ); }
    return *this;
  }

  StreamBuffer& operator<<( const std::string& data ) {
    if ( 0 == m_analyzer ) {
      const char* ptr = data.c_str();
      long        len = data.length();
      writeBytes( ptr, len );
    } else {
      STREAM_ANALYSE( data, sizeof( data ) );
    }
    return *this;
  }

  template <class TYPE>
  StreamBuffer& operator>>( TYPE*& refpObject ) {
    return getObjectPointer( refpObject, refpObject );
  }

  StreamBuffer& operator<<( const ContainedObject* pObject ) {
    STREAM_ANALYSE( pObject, sizeof( pObject ) );
    addContainedLink( pObject, INVALID, INVALID );
    return *this;
  }

  StreamBuffer& operator<<( const DataObject* pObject ) {
    STREAM_ANALYSE( pObject, sizeof( pObject ) );
    addIdentifiedLink( pObject, INVALID );
    return *this;
  }

  void serialize( DataIO& ioObject ) {
    ioObject.serialize( *this );
    m_pointer = 0;
  }
};
 
#undef STREAM_ANALYSE

inline StreamBuffer::SwapAction StreamBuffer::swapBuffer( int siz ) const {
  switch ( siz ) {
  case 1:
    return SINGLE_BYTE;
  default:
#if defined( __alpha ) && !defined( __VMS )
    //    return m_swapEnabled ? SWAP : NOSWAP;
    return NOSWAP;
#elif defined( __sun ) && defined( __SVR4 ) && defined( __i386 )
    //    return m_swapEnabled ? SWAP : NOSWAP;
    return NOSWAP;
#elif defined( __APPLE__ )
    //    return m_swapEnabled ? SWAP : NOSWAP;
    return SWAP;
#elif defined( __linux ) && !defined( __powerpc )
    //    return m_swapEnabled ? SWAP : NOSWAP;
    return NOSWAP;
#else
    return m_swapEnabled ? SWAP : NOSWAP;
//    return NOSWAP;
#endif
  }
}

inline void StreamBuffer::swapToBuffer( const void* source, int siz ) {
  char buff[8], *tar, *src = (char*)source;
  extend( m_pointer + siz );
  tar = m_buffer + m_pointer;
  switch ( swapBuffer( siz ) ) {
  case SINGLE_BYTE:
    *tar = *src;
    break;
  case SWAP:
#ifdef __APPLE__
    for ( int i = 0, j = siz - 1; i < siz; i++, j-- ) tar[j] = src[i];
#else
    ::swab( src, buff, siz );
#endif
    src = buff;
    [[fallthrough]];
  case NOSWAP:
    std::copy_n( src, siz, tar );
    break;
  default:
    break;
  }
  m_pointer += siz;
}

inline void StreamBuffer::swapFromBuffer( void* target, int siz ) {
  char* tar = (char*)target;
  char* src = m_buffer + m_pointer;
  switch ( swapBuffer( siz ) ) {
  case SINGLE_BYTE:
    *tar = *src;
    break;
  case SWAP:
#ifdef __APPLE__
    for ( int i = 0, j = siz - 1; i < siz; i++, j-- ) tar[j] = src[i];
#else
    ::swab( src, tar, siz );
#endif
    break;
  case NOSWAP:
    std::copy_n( src, siz, tar );
    break;
  default:
    break;
  }
  m_pointer += siz;
}
 
// Output serialize a vector of items
template <class T>
inline StreamBuffer& operator<<( StreamBuffer& s, const std::vector<T>& v ) {
  s << v.size();
  for ( const auto& i : v ) s << i;
  return s;
}
 
// Input serialize a vector of items
template <class T>
inline StreamBuffer& operator>>( StreamBuffer& s, std::vector<T>& v ) {
  long i, len;
  s >> len;
  v.clear();
  for ( i = 0; i < len; i++ ) {
    T temp;
    s >> temp;
    v.push_back( temp );
  }
  return s;
}
 
// Output serialize a list of items
template <class T>
inline StreamBuffer& operator<<( StreamBuffer& s, const std::list<T>& l ) {
  s << l.size();
  for ( const auto& i : l ) s << i;
  return s;
}
 
// Input serialize a list of items
template <class T>
inline StreamBuffer& operator>>( StreamBuffer& s, std::list<T>& l ) {
  long len;
  s >> len;
  l.clear();
  for ( long i = 0; i < len; i++ ) {
    T temp;
    s >> temp;
    l.push_back( temp );
  }
  return s;
}