// **********************************************************************
//
// Copyright (c) 2002
// IONA Technologies, Inc.
// Waltham, MA, USA
//
// All Rights Reserved
//
// **********************************************************************

#ifndef OB_STREAM_H
#define OB_STREAM_H

#include <OB/Stream_fwd.h>
#include <OB/Principal_fwd.h>
#include <OB/Object_fwd.h>
#include <OB/ValueBase_fwd.h>
#include <OB/AbstractBase_fwd.h>
#include <OB/TypeCodeConst_fwd.h>
#include <OB/TypeCode_fwd.h>
#include <OB/Custom_fwd.h>
#include <OB/ORBInstance_fwd.h>
#include <OB/OCIBuffer_fwd.h>
#include <OB/ValueReaderWriter_fwd.h>

#include <OB/Any.h>
#include <OB/OCITypes.h>
#include <OB/CodeConverters.h>
#include <OB/Hashtable.h>

namespace OB
{
//
// Forward declaration
//
class TypeCodeCache;

//
// The InputStreamImpl class (not reference counted)
//
class InputStreamImpl
{
    //
    // Hide copy-constructor and assignment operator
    //
    InputStreamImpl(const InputStreamImpl&);
    InputStreamImpl& operator=(const InputStreamImpl&);

protected:

    ORBInstance_var orbInstance_;
    OCI::BufferImpl* buf_;
    CORBA::Octet* origCur_;
    bool origSwap_;
    bool swap_;
    CodeConverters codeConverters_;
    CORBA::UShort GIOPVersion_;
    ValueReader_var valueReader_;
    bool charConversionRequired_;
    bool wCharConversionRequired_;
    TypeCodeCache* cache_;

    //
    // Note: It is assumed that if a char writer is required, the
    // value to be written has to be converted. A char writer is only
    // required for UTF-8 and in this case conversion is always required
    // since we won't ever have UTF-8 as *native* code set.
    //
    bool charReaderRequired_;
    bool charReaderOrConversionRequired_;

    struct OctetPtrHasher
    {
	static CORBA::ULong
	hash(const CORBA::Octet* k)
	{
	    return (CORBA::ULong)((long)k);
	}

	static bool
	comp(const CORBA::Octet* k1, const CORBA::Octet* k2)
	{
	    return k1 == k2;
	}
    };
    typedef Hashtable<const CORBA::Octet*, CORBA::TypeCode_ptr,
	OctetPtrHasher> ReadTCHistory;

    CORBA::TypeCode_ptr checkCache(const char*, CORBA::Octet*, CORBA::ULong);
    
    CORBA::TypeCode_ptr readTypeCodeImpl(ReadTCHistory&, bool = false);
    ValueReader_ptr valueReader();
    void checkChunk();

    friend class OutputStreamImpl;

public:

    InputStreamImpl(OCI::BufferImpl*, CORBA::ULong, bool,
		    const CodeConverters* = 0, CORBA::UShort = 0);
    virtual ~InputStreamImpl();

    void _OB_codeConverters(const CodeConverters*, CORBA::UShort);
    const CodeConverters* _OB_codeConverters()
    { return &codeConverters_; }

    OCI::BufferImpl* _OB_bufferImpl() { return buf_; }

    CORBA::ULong _OB_pos() { return buf_ -> pos(); }
    void _OB_pos(CORBA::ULong pos) { buf_ -> pos(pos); }

    bool _OB_swap() const { return swap_; }
    void _OB_swap(bool s) { swap_ = s; }

    void _OB_reset();

    void _OB_skip(CORBA::ULong length)
    {
        register CORBA::Octet* newCur = buf_ -> cur_ + length;
        if(newCur < buf_ -> cur_ || newCur > buf_ -> last_)
            throw CORBA::MARSHAL(MinorReadOverflow, CORBA::COMPLETED_NO);
        buf_ -> cur_ = newCur;
    }

    void _OB_skipAlign(CORBA::ULong align)
    {
        register CORBA::Octet* newCur = buf_ -> cur_ + align - 1;
        newCur -= (unsigned long)newCur % align;
        if(newCur < buf_ -> cur_ || newCur > buf_ -> last_)
            throw CORBA::MARSHAL(MinorReadOverflow, CORBA::COMPLETED_NO);
        buf_ -> cur_ = newCur;
    }

    void _OB_ORBInstance(ORBInstance_ptr);
    ORBInstance_ptr _OB_ORBInstance();

    void _OB_readEndian();

    CORBA::Long _OB_readLongUnchecked();

    void _OB_beginValue();
    void _OB_endValue();
    void _OB_remarshalValue(CORBA::TypeCode_ptr, OutputStreamImpl*);

    CORBA::DataInputStream* _OB_createDataInputStream();

    CORBA::Boolean read_boolean()
    {
        if(valueReader_.in())
            checkChunk();

        if(buf_ -> cur_ > buf_ -> last_)
            throw CORBA::MARSHAL(MinorReadBooleanOverflow,
                                 CORBA::COMPLETED_NO);

        CORBA::Boolean val = (*(buf_ -> cur_)++) ? true : false;
        return val;
    }

    CORBA::Octet read_octet()
    {
        if(valueReader_.in())
            checkChunk();

        if(buf_ -> cur_ > buf_ -> last_)
            throw CORBA::MARSHAL(MinorReadOctetOverflow, CORBA::COMPLETED_NO);

        CORBA::Octet val = *(buf_ -> cur_)++;
        return val;
    }

    CORBA::Short read_short();
    CORBA::UShort read_ushort();
    CORBA::Long read_long();
    CORBA::ULong read_ulong();
    CORBA::LongLong read_longlong();
    CORBA::ULongLong read_ulonglong();
    CORBA::Float read_float();
    CORBA::Double read_double();
    CORBA::LongDouble read_longdouble();
    CORBA::Char read_char();
    void read_char_array(CORBA::Char*, CORBA::ULong);
    CORBA::WChar read_wchar();
    char* read_string();
    wchar_t* read_wstring();
    CORBA::Object_ptr read_Object();
    CORBA::TypeCode_ptr read_TypeCode();
    CORBA::Principal_ptr read_Principal();
    CORBA::Any* read_any();
    void read_any(CORBA::Any&);
    CORBA::Fixed read_fixed();

    //
    // The string argument should be supplied except in special cases
    // (e.g., custom marshalling)
    //
    CORBA::ValueBase* read_value(const char* = 0);

    void read_value(CORBA::Any&, CORBA::TypeCode_ptr);

    CORBA::AbstractBase_ptr read_abstract_interface();

    void read_boolean_array(CORBA::Boolean* value, CORBA::ULong length)
    {
	if(length)
	{
	    assert_nca(value != 0, NCANullValue);
	
	    if(valueReader_.in())
		checkChunk();
	
	    register CORBA::Octet* nextCur = buf_ -> cur_ + length;
	    if(nextCur < buf_ -> cur_ || nextCur > buf_ -> last_)
		throw CORBA::MARSHAL(MinorReadBooleanArrayOverflow,
				     CORBA::COMPLETED_NO);

	    memcpy(value, buf_ -> cur_, length);
	    buf_ -> cur_ = nextCur;
	}
    }

    void read_octet_array(CORBA::Octet* value, CORBA::ULong length)
    {
        if(length)
        {
            assert_nca(value != 0, NCANullValue);

            if(valueReader_.in())
                checkChunk();

            register CORBA::Octet* nextCur = buf_ -> cur_ + length;
            if(nextCur < buf_ -> cur_ || nextCur > buf_ -> last_)
                throw CORBA::MARSHAL(MinorReadOctetArrayOverflow,
                                     CORBA::COMPLETED_NO);

            memcpy(value, buf_ -> cur_, length);
            buf_ -> cur_ = nextCur;
        }
    }

    void read_wchar_array(CORBA::WChar*, CORBA::ULong);
    void read_short_array(CORBA::Short*, CORBA::ULong);
    void read_ushort_array(CORBA::UShort*, CORBA::ULong);
    void read_long_array(CORBA::Long*, CORBA::ULong);
    void read_ulong_array(CORBA::ULong*, CORBA::ULong);
    void read_longlong_array(CORBA::LongLong*, CORBA::ULong);
    void read_ulonglong_array(CORBA::ULongLong*, CORBA::ULong);
    void read_float_array(CORBA::Float*, CORBA::ULong);
    void read_double_array(CORBA::Double*, CORBA::ULong);
    void read_longdouble_array(CORBA::LongDouble*, CORBA::ULong);

    void read_any_NoTypeCode(CORBA::Any&, CORBA::TypeCode_ptr);
};

//
// The OutputStreamImpl class (not reference counted)
//
class OutputStreamImpl
{
    //
    // Hide copy-constructor and assignment operator
    //
    OutputStreamImpl(const OutputStreamImpl&);
    OutputStreamImpl& operator=(const OutputStreamImpl&);

protected:

    OCI::BufferImpl* buf_;
    CodeConverters codeConverters_;
    CORBA::UShort GIOPVersion_;
    bool charConversionRequired_;
    bool wCharConversionRequired_;

    //
    // Note: It is assumed that if a char writer is required, the
    // value to be written has to be converted. A char writer is only
    // required for UTF-8 and in this case conversion is always required
    // since we won't ever have UTF-8 as *native* code set.
    //
    bool charWriterRequired_;
    bool charWriterOrConversionRequired_;

    //
    // Handles all OBV marshalling
    //
    ValueWriter_var valueWriter_;

    //
    // If alignNext_ > 0, the next write should be aligned on this
    // boundary
    //
    CORBA::ULong alignNext_;

    struct TypeCodePtrHasher
    {
	static CORBA::ULong
	hash(const CORBA::TypeCode_ptr k)
	{
	    return (CORBA::ULong)((long)k);
	}

	static bool
	comp(const CORBA::TypeCode_ptr k1, const CORBA::TypeCode_ptr k2)
	{
	    return k1 == k2;
	}
    };
    typedef Hashtable<CORBA::TypeCode_ptr, CORBA::Long,
	TypeCodePtrHasher> WriteTCHistory;

    CORBA::ULong writeGap();
    void writeLength(CORBA::ULong);
    void writeTypeCodeImpl(CORBA::TypeCode_ptr, WriteTCHistory&);

    void checkBeginChunk();
    ValueWriter_ptr valueWriter();

    void addCapacity(CORBA::ULong size)
    {
        //
        // Expand buffer to hold requested size
        //
        // Note: OutputStreamImpls are not always written to in a
        //       linear fashion, i.e., sometimes the position is reset
        //       to an earlier point and data is patched
        //       in. Therefore, do NOT do this:
        //
        //buf_ -> realloc(buf_ -> len_ + size);
        //
        // 
        if(alignNext_ > 0)
        {
            CORBA::ULong align = alignNext_;
            alignNext_ = 0;
            addCapacity(size, align);
        }
        else
        {
            //
            // If we're at the end of the current buffer, then we are about
            // to write new data. We must first check if we need to start a
            // chunk, which may result in a recursive call to addCapacity().
            //
            if(buf_ -> cur_ == buf_ -> last_ && valueWriter_.in())
                checkBeginChunk();

            //
            // If there isn't enough room, then reallocate the buffer
            //
            CORBA::ULong len = buf_ -> cur_ - buf_ -> data_ + size;
            if(len > buf_ -> len_)
                buf_ -> realloc(len);
        }
    }

    void addCapacity(CORBA::ULong size, CORBA::ULong align)
    {
        assert(align > 0); // use addCapacity(ULong) if align == 0

        //
        // If we're at the end of the current buffer, then we are about
        // to write new data. We must first check if we need to start a
        // chunk, which may result in a recursive call to addCapacity().
        //
        if(buf_ -> cur_ == buf_ -> last_ && valueWriter_.in())
            checkBeginChunk();

        //
        // If alignNext_ is set, then use the larger of alignNext_ and align
        //
        if(alignNext_ > 0)
        {
            align = (alignNext_ > align ? alignNext_ : align);
            alignNext_ = 0;
        }

        //
        // Align to the requested boundary
        //
        register CORBA::Octet* newCur = buf_ -> cur_ + align - 1;
        newCur -= CORBA::ULong((long)newCur) % align;
        buf_ -> cur_ = newCur;

        //
        // If there isn't enough room, then reallocate the buffer
        //
        if(newCur + size > buf_ -> last_)
            buf_ -> realloc(newCur - buf_ -> data_ + size);
    }

    friend class InputStreamImpl;

public:

    OutputStreamImpl(OCI::BufferImpl*, const CodeConverters* = 0,
		     CORBA::UShort = 0);
    virtual ~OutputStreamImpl();

    OCI::BufferImpl* _OB_bufferImpl() { return buf_; }

    CORBA::ULong _OB_pos() { return buf_ -> pos(); }
    void _OB_pos(CORBA::ULong pos) { buf_ -> pos(pos); }

    void _OB_align(CORBA::ULong n)
    {
        if(CORBA::ULong((long)(buf_ -> cur_)) % n != 0)
            addCapacity(0, n);
    }

    void _OB_alignNext(CORBA::ULong n)
    {
        alignNext_ = n;
    }

    void _OB_beginValue(CORBA::Long, const OB::StrSeq< int >&, bool);
    void _OB_endValue();

    CORBA::DataOutputStream* _OB_createDataOutputStream();

    void _OB_writeEndian() { write_boolean(Endian); }

    InputStream_ptr create_input_stream();

    void write_boolean(CORBA::Boolean value)
    {
        addCapacity(1);
        *(buf_ -> cur_) = value;
        buf_ -> cur_ += 1;
    }

    void write_octet(CORBA::Octet value)
    {
        addCapacity(1);
        *(buf_ -> cur_) = value;
        buf_ -> cur_ += 1;
    }

    void write_short(CORBA::Short value)
    {
        addCapacity(2, 2);
        *(CORBA::Short*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 2;
    }

    void write_ushort(CORBA::UShort value)
    {
        addCapacity(2, 2);
        *(CORBA::UShort*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 2;
    }

    void write_long(CORBA::Long value)
    {
        addCapacity(4, 4);
        *(CORBA::Long*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 4;
    }

    void write_ulong(CORBA::ULong value)
    {
        addCapacity(4, 4);
        *(CORBA::ULong*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 4;
    }

    void write_longlong(CORBA::LongLong value)
    {
        addCapacity(8, 8);
        *(CORBA::LongLong*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 8;
    }

    void write_ulonglong(CORBA::ULongLong value)
    {
        addCapacity(8, 8);
        *(CORBA::ULongLong*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 8;
    }

    void write_float(CORBA::Float value)
    {
        addCapacity(4, 4);
        *(CORBA::Float*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 4;
    }

    void write_double(CORBA::Double value)
    {
        addCapacity(8, 8);
        *(CORBA::Double*)(buf_ -> cur_) = value;
        buf_ -> cur_ += 8;
    }

    void write_longdouble(CORBA::LongDouble);
    void write_char(CORBA::Char);
    void write_char_array(const CORBA::Char*, CORBA::ULong);
    void write_wchar(CORBA::WChar);
    void write_string(const char*);
    void write_wstring(const wchar_t*);
    void write_Object(CORBA::Object_ptr);
    void write_TypeCode(CORBA::TypeCode_ptr);
    void write_Principal(CORBA::Principal_ptr);
    void write_any(const CORBA::Any&);
    void write_fixed(const CORBA::Fixed&);

    //
    // The string argument should be supplied except in special cases
    // (e.g., custom marshalling).
    //
    void write_value(CORBA::ValueBase*, const char* = 0);
    void write_abstract_interface(CORBA::AbstractBase_ptr);

    void write_boolean_array(const CORBA::Boolean*, CORBA::ULong);
    void write_octet_array(const CORBA::Octet*, CORBA::ULong);
    void write_short_array(const CORBA::Short*, CORBA::ULong);
    void write_ushort_array(const CORBA::UShort*, CORBA::ULong);
    void write_long_array(const CORBA::Long*, CORBA::ULong);
    void write_ulong_array(const CORBA::ULong*, CORBA::ULong);
    void write_longlong_array(const CORBA::LongLong*, CORBA::ULong);
    void write_ulonglong_array(const CORBA::ULongLong*, CORBA::ULong);
    void write_float_array(const CORBA::Float*, CORBA::ULong);
    void write_double_array(const CORBA::Double*, CORBA::ULong);

    void write_wchar_array(const CORBA::WChar*, CORBA::ULong);
    void write_longdouble_array(const CORBA::LongDouble*, CORBA::ULong);

    void write_any_NoTypeCode(const CORBA::Any&);
    void write_InputStreamImpl(InputStreamImpl*, CORBA::TypeCode_ptr);
};

//
// The InputStream class (reference counted)
//
class InputStream : public InputStreamImpl, public SimpleRefCount
{
    OCI::Buffer_var bufVar_;

public:

    InputStream(OCI::Buffer_ptr, CORBA::ULong, bool, const CodeConverters* = 0,
		CORBA::UShort = 0);
    virtual ~InputStream();

    static inline InputStream_ptr _duplicate(InputStream_ptr p)
    { if(p) p -> _OB_incRef(); return p; }
    static inline InputStream_ptr _nil()
    { return 0; }

    InputStream_ptr _OB_clone();
    OCI::Buffer_ptr _OB_buffer_nodup() { return bufVar_.in(); }
    OCI::Buffer_ptr _OB_buffer_retn() { return bufVar_._retn(); }
};

//
// The OutputStream class (reference counted)
//
class OutputStream : public OutputStreamImpl, public SimpleRefCount
{
    OCI::Buffer_var bufVar_;

public:

    OutputStream(OCI::Buffer_ptr, const CodeConverters* = 0,
		 CORBA::UShort = 0);
    virtual ~OutputStream();

    static inline OutputStream_ptr _duplicate(OutputStream_ptr p)
    { if(p) p -> _OB_incRef(); return p; }
    static inline OutputStream_ptr _nil()
    { return 0; }

    OCI::Buffer_ptr _OB_buffer_nodup() { return bufVar_.in(); }
    OCI::Buffer_ptr _OB_buffer_retn() { return bufVar_._retn(); }
};

} // End of namespace OB

namespace CORBA
{

inline void
release(OB::InputStream_ptr p)
{
    if(p)
        p -> _OB_decRef();
}

inline Boolean
is_nil(OB::InputStream_ptr p)
{
    return p == 0;
}

inline void
release(OB::OutputStream_ptr p)
{
    if(p)
        p -> _OB_decRef();
}

inline Boolean
is_nil(OB::OutputStream_ptr p)
{
    return p == 0;
}

} // End of namespace CORBA

#endif
