streambuf.h

Go to the documentation of this file.

 
 
#pragma once
 
#include <cx/buffer/bufring.h>
#include <cx/stype/stype.h>
 
typedef struct StreamBuffer StreamBuffer;
 
// Pull callback
// sz is set to the maximum amount of data the may be written to buf.
// The callback should fill buf with up to that amount and return the number
// of bytes that were written.
// The callback may return 0 if no data is currently available but will likely
// be immediately called again, so a performing a blocking wait is advisable.
//
// If the callback needs to write an amount of data that is larger than the
// space available as indicated by sz, it may instead call sbufPWrite with the
// full amount (which will expand the buffer in the process) and return 0.
//
// If sz is 0, check if the consumer is finished and/or for error state.
typedef size_t (*sbufPullCB)(_Pre_valid_ StreamBuffer* sb, _Out_writes_bytes_(sz) uint8* buf,
                             size_t sz, _Pre_opt_valid_ void* ctx);
 
// Push callback
// When this callback is used (in direct mode), the data is pushed once to the
// callback and MUST all be written in one go or it will be lost.
// If sz is 0, check if the producer is finished and/or for error state.
typedef void (*sbufPushCB)(_Pre_valid_ StreamBuffer* sb, _In_reads_bytes_(sz) const uint8* buf,
                           size_t sz, _Pre_opt_valid_ void* ctx);
 
// Send callback
// This callback is used with sbufCSend. It may be called multiple times with varying
// offsets. The offset passed is always from the start of the available bytes in the
// buffer.
// If this function returns true, the sent bytes will be consumed and removed from the
// buffer. If it returns false, it behaves like the peek functions and does not remove
// the bytes from the buffer.
// If sz is 0, check if the producer is finished and/or for error state.
typedef bool (*sbufSendCB)(_Pre_valid_ StreamBuffer* sb, _In_reads_bytes_(sz) const uint8* buf,
                           size_t off, size_t sz, _Pre_opt_valid_ void* ctx);
 
// Notify callback
// Notification to a consumer that data is available. The sbufC* functions may be
// used to read all or part of the available data.
// If sz is 0, check if the producer is finished and/or for error state.
typedef void (*sbufNotifyCB)(_Pre_valid_ StreamBuffer* sb, size_t sz, _Pre_opt_valid_ void* ctx);
 
// Cleanup callback
// Called just before the structure is deallocated, should perform any needed
// cleanup of the user-supplied ctx.
typedef void (*sbufCleanupCB)(_Pre_opt_valid_ void* ctx);
 
 
enum STREAM_BUFFER_FLAGS_ENUM {
    SBUF_Push                = 0x0001,
    SBUF_Pull                = 0x0002,
    SBUF_Direct              = 0x0010,
    SBUF_Error               = 0x0800,
    SBUF_Producer_Registered = 0x1000,
    SBUF_Consumer_Registered = 0x2000,
    SBUF_Producer_Done       = 0x4000,
    SBUF_Consumer_Done       = 0x8000,
};
 
typedef struct StreamBuffer {
    BufRing buf;                     
    size_t targetsz;                 
 
    sbufPullCB producerPull;         
    sbufCleanupCB producerCleanup;   
    void* producerCtx;               
 
    sbufNotifyCB consumerNotify;     
    sbufPushCB consumerPush;         
    sbufCleanupCB consumerCleanup;   
    void* consumerCtx;               
 
    int refcount;                    
    uint32 flags;                    
} StreamBuffer;
 
_Ret_valid_ StreamBuffer* sbufCreate(size_t targetsz);
 
_At_(*sb, _Pre_maybenull_ _Post_null_) void sbufRelease(_Inout_ StreamBuffer** sb);
 
void sbufError(_Inout_ StreamBuffer* sb);
 
_meta_inline bool sbufIsPull(_In_ StreamBuffer* sb)
{
    return sb->flags & SBUF_Pull;
}
 
_meta_inline bool sbufIsPush(_In_ StreamBuffer* sb)
{
    return sb->flags & SBUF_Push;
}
 
_meta_inline bool sbufIsError(_In_ StreamBuffer* sb)
{
    return sb->flags & SBUF_Error;
}
 
_meta_inline bool sbufIsPFinished(_In_ StreamBuffer* sb)
{
    return (sb->flags & SBUF_Producer_Done) || sbufIsError(sb);
}
 
_meta_inline bool sbufIsCFinished(_In_ StreamBuffer* sb)
{
    return (sb->flags & SBUF_Consumer_Done) || sbufIsError(sb);
}
 
 
 
_Check_return_ bool sbufPRegisterPull(_Inout_ StreamBuffer* sb, _In_ sbufPullCB ppull,
                                      _In_opt_ sbufCleanupCB pcleanup, _Inout_opt_ void* ctx);
 
_Check_return_ bool sbufPRegisterPush(_Inout_ StreamBuffer* sb, _In_opt_ sbufCleanupCB pcleanup,
                                      _Inout_opt_ void* ctx);
 
size_t sbufPAvail(_In_ StreamBuffer* sb);
 
bool sbufPWrite(_Inout_ StreamBuffer* sb, _In_reads_bytes_(sz) const uint8* buf, size_t sz);
 
bool sbufPWriteStr(_Inout_ StreamBuffer* sb, _In_opt_ strref str);
 
bool sbufPWriteLine(_Inout_ StreamBuffer* sb, _In_opt_ strref str);
 
bool sbufPWriteEOL(_Inout_ StreamBuffer* sb);
 
void sbufPFinish(_Pre_valid_ _Post_invalid_ StreamBuffer* sb);
 
 
 
_Check_return_ bool sbufCRegisterPull(_Inout_ StreamBuffer* sb, _In_opt_ sbufCleanupCB ccleanup,
                                      _Inout_opt_ void* ctx);
 
_Check_return_ bool sbufCRegisterPush(_Inout_ StreamBuffer* sb, _In_ sbufNotifyCB cnotify,
                                      _In_opt_ sbufCleanupCB ccleanup, _Inout_opt_ void* ctx);
 
_Check_return_ bool sbufCRegisterPushDirect(_Inout_ StreamBuffer* sb, _In_ sbufPushCB cpush,
                                            _In_opt_ sbufCleanupCB ccleanup, _Inout_opt_ void* ctx);
 
size_t sbufCAvail(_Inout_ StreamBuffer* sb);
 
_Success_(return) bool
sbufCRead(_Inout_ StreamBuffer* sb, _Out_writes_bytes_to_(sz, *bytesread) uint8* buf, size_t sz,
          _Out_ _Deref_out_range_(0, sz) size_t* bytesread);
 
_Success_(return > 0) bool
sbufCPeek(_Inout_ StreamBuffer* sb, _Out_writes_bytes_(sz) uint8* buf, size_t off, size_t sz);
 
bool sbufCFeed(_Inout_ StreamBuffer* sb, size_t minsz);
 
bool sbufCSend(_Inout_ StreamBuffer* sb, _In_ sbufSendCB func, size_t sz);
 
bool sbufCSkip(_Inout_ StreamBuffer* sb, size_t bytes);
 
void sbufCFinish(_Pre_valid_ _Post_invalid_ StreamBuffer* sb);