tbox中提供了常用的一些stream模块，例如：data、file、http、sock等，可以通过指定不同的url，使用相同的接口 进行数据流的读写，非常的方便。

    例如：

    1.     // init stream
    2.     tb_stream_ref_t stream = tb_stream_init_from_url("http://www.xxx.com/file.txt");
    3.     if (stream)
    4.     {
    5.         // open stream
    6.         if (tb_stream_open(stream))
    7.         {
    8.             // read line
    9.             tb_long_t size = 0;
    10.             tb_char_t line[TB_STREAM_BLOCK_MAXN];
    11.             while ((size = tb_stream_bread_line(stream, line, sizeof(line))) >= 0)
    12.             {
    13.                 // trace
    14.                 tb_trace_i("line: %s", line);
    15.             }
    16.         }
    18.         // exit stream
    19.         tb_stream_exit(stream);
    20.     }

    这样的好处是，操作io的模块不需要关心实际的数据流协议，只管从stream中读写数据就行了，实现数据和业务逻辑的解耦。。

    但是tbox提供的这些内置stream模块，有时候没法完全咱们的实际需求，例如：

    我想读取一个实时数据流的缓存队列，这个数据流一段会不停的送入数据进来，另外一段会不停的读取数据，如果数据不够，就会进入等待

    这其实是个很有用的功能，我的很多需求都会用到，例如：流媒体的一些实时数据获取和复用等等。。

    那如何实现这样一个stream模块，让tbox的stream接口支持呢，我们只要实现一个自定义的流模块就好，实现起来也不复杂

    我们先定义个一个stream类型，例如：

    1.     // 用户自定义流类型：实时流
    2.     #define TB_STREAM_TYPE_REAL         (TB_STREAM_TYPE_USER + 1)
    4.     // 定义一个控制流代码，之后tb_stream_ctrl需要
    5.     #define TM_STREAM_CTRL_REAL_PUSH    TB_STREAM_CTRL(TM_STREAM_TYPE_REAL, 1)

    定义个自定义流的数据结构，用于维护咱们的私有数据

    1.     // 实时流类型
    2.     typedef struct __tb_stream_real_t
    3.     {
    4.         // 这里定义了一个数据块buffer的队列，用于缓存不断送入的数据
    5.         tb_queue_ref_t      buffers;
    7.         // 总的数据大小
    8.         tb_size_t           size;
    10.     }tb_stream_real_t, *tb_stream_real_ref_t;
    12.     // 定义一个buffer块类型，用于维护单个数据块
    13.     typedef struct __tm_real_buffer_t
    14.     {
    15.         // 数据地址
    16.         tb_byte_t*          data;
    18.         // 这个buffer总大小
    19.         tb_size_t           size;
    21.         // 在这个buffer中，当前读取到的数据
    22.         tb_size_t           read;
    24.     }tm_real_buffer_t, *tm_real_buffer_ref_t;

    创建一个stream实例，注册一些需要的回调接口

    1.     // 初始化创建个一个实时流
    2.     tb_stream_ref_t tb_stream_init_real()
    3.     {
    4.         return tb_stream_init(  TB_STREAM_TYPE_REAL
    5.                             ,   sizeof(tb_stream_real_t)
    6.                             ,   0           // stream缓存大小（file/sock有用），这里禁用了，因为咱们的流不需要缓存读取
    7.                             ,   tb_stream_real_open
    8.                             ,   tb_stream_real_clos
    9.                             ,   tb_stream_real_exit
    10.                             ,   tb_stream_real_ctrl
    11.                             ,   tb_stream_real_wait
    12.                             ,   tb_stream_real_read
    13.                             ,   tb_null     // 写回调，这里不需要
    14.                             ,   tb_null     // seek，我们这里不需要
    15.                             ,   tb_null     // 刷新写数据，不需要
    16.                             ,   tb_null);   // kill当前的stream，很少用，一般用于中断内部读写
    17.     }

    下面就是具体的回调接口实现了

    1.     // 实现open回调接口，用于打开stream，tb_stream_open会用到
    2.     static tb_bool_t tb_stream_real_open(tb_stream_ref_t stream)
    3.     {
    4.         // check
    5.         tb_stream_real_ref_t rstream = (tb_stream_real_ref_t)stream;
    6.         tb_assert_and_check_return_val(rstream, tb_false);
    8.         // 初始化一个buffer队列，并注册自动释放接口：tb_real_buffer_exit，之后有说明
    9.         rstream->buffers = tb_queue_init(0, tb_element_mem(sizeof(tb_real_buffer_t), tb_real_buffer_exit, tb_null));
    11.         // init size
    12.         rstream->size = 0;
    14.         // ok
    15.         return !!rstream->buffers;
    16.     }
    18.     // 实现close回调接口，用于关闭stream，tb_stream_clos会用到
    19.     static tb_bool_t tb_stream_real_clos(tb_stream_ref_t stream)
    20.     {
    21.         // check
    22.         tb_stream_real_ref_t rstream = (tb_stream_real_ref_t)stream;
    23.         tb_assert_and_check_return_val(rstream, tb_false);
    25.         // exit buffers
    26.         if (rstream->buffers) tb_queue_exit(rstream->buffers);
    27.         rstream->buffers = tb_null;
    29.         // ok
    30.         return tb_true;
    31.     }
    33.     // 实现exit回调接口，用于销毁stream，tb_stream_exit会用到
    34.     static tb_void_t tb_stream_real_exit(tb_stream_ref_t stream)
    35.     {
    36.         // check
    37.         tb_stream_real_ref_t rstream = (tb_stream_real_ref_t)stream;
    38.         tb_assert_and_check_return(rstream);
    40.         // exit buffers
    41.         if (rstream->buffers) tb_queue_exit(rstream->buffers);
    42.         rstream->buffers = tb_null;
    44.         // clear size
    45.         rstream->size = 0;
    46.     }
    48.     // 实现read回调接口，用于读取数据，tb_stream_read/tb_stream_bread等接口会用到
    49.     static tb_long_t tb_stream_real_read(tb_stream_ref_t stream, tb_byte_t* data, tb_size_t size)
    50.     {
    51.         // check
    52.         tb_stream_real_ref_t rstream = (tb_stream_real_ref_t)stream;
    53.         tb_assert_and_check_return_val(rstream && rstream->buffers, -1);
    55.         // check
    56.         tb_check_return_val(data, -1);
    57.         tb_check_return_val(size, 0);
    59.         // 依次从队列头部读取每块buffer的数据，直到读满为止
    60.         tb_long_t read = 0;
    61.         while (read < size && tb_queue_size(rstream->buffers))
    62.         {
    63.             // get buffer
    64.             tb_real_buffer_ref_t buffer = tb_queue_get(rstream->buffers);
    65.             tb_assert_and_check_break(buffer && buffer->data && buffer->size);
    67.             // read data
    68.             if (buffer->read < buffer->size)
    69.             {
    70.                 // calculate the need size
    71.                 tb_size_t need = tb_min(size - read, buffer->size - buffer->read);
    73.                 // copy data
    74.                 tb_memcpy(data + read, buffer->data + buffer->read, need);
    76.                 // update the read size for buffer
    77.                 buffer->read += need;
    79.                 // update the total read size
    80.                 read += need;
    81.             }
    83.             // 将读空的buffer释放掉
    84.             if (buffer->read == buffer->size)
    85.                 tb_queue_pop(rstream->buffers);
    86.         }
    88.         // ok?
    89.         return read;
    90.     }
    92.     // 实现wait回调接口，用于等待数据，tb_stream_wait/tb_stream_bread等阻塞读取接口会用到
    93.     static tb_long_t tb_stream_real_wait(tb_stream_ref_t stream, tb_size_t wait, tb_long_t timeout)
    94.     {
    95.         // check
    96.         tb_stream_real_ref_t rstream = (tb_stream_real_ref_t)stream;
    97.         tb_assert_and_check_return_val(rstream && rstream->buffers, -1);
    99.         // 当前是否有数据可读？
    100.         return tb_queue_size(rstream->buffers)? TB_STREAM_WAIT_READ : TB_STREAM_WAIT_NONE;
    101.     }
    103.     // 实现ctrl回调接口，用于设置和获取一些状态，扩展一些自定义的接口，tb_stream_ctrl接口会用到
    104.     static tb_bool_t tb_stream_real_ctrl(tb_stream_ref_t stream, tb_size_t ctrl, tb_va_list_t args)
    105.     {
    106.         // check
    107.         tb_stream_real_ref_t rstream = (tb_stream_real_ref_t)stream;
    108.         tb_assert_and_check_return_val(rstream, tb_false);
    110.         // ctrl
    111.         switch (ctrl)
    112.         {
    113.         case TB_STREAM_CTRL_GET_SIZE:
    114.             {
    115.                 // the psize
    116.                 tb_hong_t* psize = (tb_hong_t*)tb_va_arg(args, tb_hong_t*);
    117.                 tb_assert_and_check_break(psize);
    119.                 // 获取数据流大小，tb_stream_size有用到
    120.                 *psize = rstream->size;
    122.                 // ok
    123.                 return tb_true;
    124.             }   
    125.             // 在另外一端通过tb_stream_ctrl来不断的送入数据块到stream
    126.         case TB_STREAM_CTRL_REAL_PUSH:
    127.             {
    128.                 // check
    129.                 tb_assert_and_check_break(rstream->buffers);
    131.                 // the data and size
    132.                 tb_byte_t const*    data = (tb_byte_t const*)tb_va_arg(args, tb_byte_t const*);
    133.                 tb_size_t           size = (tb_size_t)tb_va_arg(args, tb_size_t);
    134.                 tb_assert_and_check_break(data && size);
    136.                 // 压入一个数据块
    137.                 tb_real_buffer_t buffer;
    138.                 buffer.data = tb_memdup(data, size);
    139.                 buffer.size = size;
    140.                 buffer.read = 0;
    141.                 tb_queue_put(rstream->buffers, &buffer);
    143.                 // 更新总的数据大小
    144.                 rstream->size += size;
    146.                 // ok
    147.                 return tb_true;
    148.             }
    149.         default:
    150.             break;
    151.         }
    153.         // failed
    154.         return tb_false;
    155.     }

    通过上面四步， 基本上一个自定义流就实现好了，上面说的tb_real_buffer_exit主要用于queue维护的buffer的自动释放 详细说明和使用见容器章节，下面附属相关实现：

    1.     static tb_void_t tb_real_buffer_exit(tb_element_ref_t element, tb_pointer_t buff)
    2.     {
    3.         // check
    4.         tb_real_buffer_ref_t buffer = (tb_real_buffer_ref_t)buff;
    5.         tb_assert_and_check_return(buffer);
    7.         // exit it
    8.         if (buffer->data) tb_free(buffer->data);
    9.         buffer->data = tb_null;
    10.         buffer->size = 0;
    11.         buffer->read = 0;
    12.     }

    最后，贴下咱们这个自定义stream使用：

    接收端

    1.     // init stream
    2.     tb_stream_ref_t stream = tb_stream_init_real();
    3.     if (stream)
    4.     {
    5.         // open stream
    6.         if (tb_stream_open(stream))
    7.         {
    8.             // read line
    9.             tb_long_t size = 0;
    10.             tb_char_t line[TB_STREAM_BLOCK_MAXN];
    11.             while ((size = tb_stream_bread_line(stream, line, sizeof(line))) >= 0)
    12.             {
    13.                 // trace
    14.                 tb_trace_i("line: %s", line);
    15.             }
    16.         }
    18.         // exit stream
    19.         tb_stream_exit(stream);
    20.     }

    基本上没什么变化，就是换了下stream的初始化创建接口

    输入端

    1.     // 将数据不停的送入stream中
    2.     while (1)
    3.     {
    4.         // fill data
    5.         tb_byte_t data[8192];
    6.         tb_memset(data, 0xff, sizeof(data));
    8.         // push data
    9.         tb_stream_ctrl(stream, TB_STREAM_CTRL_REAL_PUSH, data, sizeof(data));
    10.     }

    上面介绍的实现和使用方式，只是个例子，方便理解tbox中stream的机制，具体实现和使用还是需要根据自己的实际需求做调整。

    更详细的使用和扩展，可参考源代码来了解。。