流模塊單獨分出來講是因為內(nèi)容相對比較多，而且也有一定難度。流模塊可以對應(yīng)數(shù)據(jù)的生產(chǎn)者/消費者模型，生產(chǎn)者可以向流里寫數(shù)據(jù)（生產(chǎn)數(shù)據(jù)），消費者從流里讀取數(shù)據(jù)（消費數(shù)據(jù)）。并且，通過回調(diào)接口，可以實現(xiàn)自動流控。VSF中的流模塊的實現(xiàn)，也可以用來闡述面向?qū)ο蟮木幊趟枷?，因為流只是一個抽象類，實際使用的是具體的fifo流、buffer流或者multibuf流等具體內(nèi)存結(jié)構(gòu)實現(xiàn)的流。所以，流只是一個標(biāo)準(zhǔn)接口，如果一個模塊支持流接口的話，就可以很方便的和其他支持流接口的模塊對接。比如，wave播放模塊的輸入流，可以接到文件流，對應(yīng)播放本地的wav文件；也可以接到http流，對用播放網(wǎng)上的wav文件，而wave播放模塊并不需要在意流的來源。

下面介紹一下流的接口，和實現(xiàn)自動流控的原理：

struct vsf_stream_cb_t
{
    void *param;
    void (*on_inout)(void *param);
    void (*on_connect)(void *param);
    void (*on_disconnect)(void *param);
};

struct vsf_stream_t
{
    // user_mem points to user structure, eg queue/fifo
    struct vsf_stream_op_t const *op;

    // callback_tx is notification for tx end of the stream
    // when rx end read the data out, will notify the tx end
    struct vsf_stream_cb_t callback_tx;
    // callback_rx is notification for rx end of the stream
    // when tx end write the data in, will notify the rx end
    struct vsf_stream_cb_t callback_rx;
    bool tx_ready;
    bool rx_ready;
    bool overflow;
};

vsf_err_t stream_init(struct vsf_stream_t *stream);
vsf_err_t stream_fini(struct vsf_stream_t *stream);
uint32_t stream_write(struct vsf_stream_t *stream, struct vsf_buffer_t *buffer);
uint32_t stream_read(struct vsf_stream_t *stream, struct vsf_buffer_t *buffer);
uint32_t stream_get_data_size(struct vsf_stream_t *stream);
uint32_t stream_get_free_size(struct vsf_stream_t *stream);
uint32_t stream_get_wbuf(struct vsf_stream_t *stream, uint8_t **ptr);
uint32_t stream_get_rbuf(struct vsf_stream_t *stream, uint8_t **ptr);
void stream_connect_rx(struct vsf_stream_t *stream);
void stream_connect_tx(struct vsf_stream_t *stream);
void stream_disconnect_rx(struct vsf_stream_t *stream);
void stream_disconnect_tx(struct vsf_stream_t *stream);

上面是流的數(shù)據(jù)結(jié)構(gòu)，和操作接口。初始化和讀寫函數(shù)一看就能明白，stream_get_data_size是得到流里的數(shù)據(jù)大小，stream_get_free_size是得到流里的空余空間大小。stream_get_wbuf和stream_get_rbuf用于得到當(dāng)前的讀寫指針，只在非常特殊的情況下使用。stream_connect_XX和stream_disconnect_XX是流的接收或者發(fā)送端的連接或者斷開。vsf_stream_t中，callback_XX是接收或者發(fā)送端設(shè)置的回調(diào)函數(shù)，當(dāng)流的一端連接/斷開/讀寫數(shù)據(jù)的時候，通過回調(diào)接口通知流的另一端。

基于callback，就可以實現(xiàn)流控，生產(chǎn)者寫入數(shù)據(jù)到流的時候，會通知消費者去消費數(shù)據(jù)；消費者消費了數(shù)據(jù)的時候，也會通知生產(chǎn)者。生產(chǎn)者可以通過stream_get_free_size來得到流里空余空間的小大，只有大于生產(chǎn)者一次可產(chǎn)生的數(shù)據(jù)的時候（比如對于全速USB，就是64字節(jié)的最大ep大小），生產(chǎn)者才會產(chǎn)生數(shù)據(jù)。如果空余空間不夠的話，生產(chǎn)者只需要等待消費者消費數(shù)據(jù)后，再次通知生產(chǎn)者，然后生產(chǎn)者再判斷是否有足夠的空余空間。當(dāng)然，實現(xiàn)流控的前提是生產(chǎn)者可以控制數(shù)據(jù)的產(chǎn)生。

流的實現(xiàn)：

uint32_t stream_read(struct vsf_stream_t *stream, struct vsf_buffer_t *buffer)
{
    uint32_t count = stream->op->read(stream, buffer);

    if (stream->tx_ready && (stream->callback_tx.on_inout != NULL) && count)
    {
        stream->callback_tx.on_inout(stream->callback_tx.param);
    }
    return count;
}

uint32_t stream_write(struct vsf_stream_t *stream, struct vsf_buffer_t *buffer)
{
    uint32_t count = stream->op->write(stream, buffer);

    if (count < buffer->size)
    {
        stream->overflow = true;
    }
    if (stream->rx_ready && (stream->callback_rx.on_inout != NULL) && count)
    {
        stream->callback_rx.on_inout(stream->callback_rx.param);
    }
    return count;
}

這里只是簡單用讀寫接口舉例，實際的讀寫操作，有op參數(shù)里指定的讀寫接口實現(xiàn)。代碼里，也只是簡單調(diào)用op指定的讀寫接口，然后判斷是否溢出以及是否需要調(diào)用回調(diào)接口。

流結(jié)構(gòu)中，并沒有指定緩沖的數(shù)據(jù)結(jié)構(gòu)，因為應(yīng)用可以根據(jù)實際應(yīng)用需求，來選擇緩沖類型。這里用最常用的fifo流來舉例：

struct vsf_fifostream_t
{
    struct vsf_stream_t stream;
    struct vsf_fifo_t mem;
};

static void fifo_stream_init(struct vsf_stream_t *stream)
{
    struct vsf_fifostream_t *fifostream = (struct vsf_fifostream_t *)stream;
    vsf_fifo_init(&fifostream->mem);
}

static uint32_t fifo_stream_get_data_length(struct vsf_stream_t *stream)
{
    struct vsf_fifostream_t *fifostream = (struct vsf_fifostream_t *)stream;
    return vsf_fifo_get_data_length(&fifostream->mem);
}

static uint32_t
fifo_stream_write(struct vsf_stream_t *stream, struct vsf_buffer_t *buffer)
{
    struct vsf_fifostream_t *fifostream = (struct vsf_fifostream_t *)stream;
    return vsf_fifo_push(&fifostream->mem, buffer->size, buffer->buffer);
}

static uint32_t
fifo_stream_read(struct vsf_stream_t *stream, struct vsf_buffer_t *buffer)
{
    struct vsf_fifostream_t *fifostream = (struct vsf_fifostream_t *)stream;
    return vsf_fifo_pop(&fifostream->mem, buffer->size, buffer->buffer);
}

const struct vsf_stream_op_t fifostream_op =
{
    .init = fifo_stream_init,
    .fini = fifo_stream_init,
    .write = fifo_stream_write,
    .read = fifo_stream_read,
    .get_data_length = fifo_stream_get_data_length,
    .get_avail_length = fifo_stream_get_avail_length,
    .get_wbuf = fifo_stream_get_wbuf,
    .get_rbuf = fifo_stream_get_rbuf,
};

這里只是列舉了幾個函數(shù)。fifo流繼承自vsf_stream_t，并且指定了fifo的緩沖類型。fifostream_op里指定了fifo流的各種操作接口，實際實現(xiàn)只是簡單調(diào)用fifo模塊的對應(yīng)接口?？梢园裿sf_fifostream_t強制類型轉(zhuǎn)換為vsf_stream_t，就可以使用標(biāo)準(zhǔn)的流接口了。實際可以通過宏來實現(xiàn)各種不同類型的流的強制轉(zhuǎn)換：

#define STREAM_INIT(s)          stream_init((struct vsf_stream_t *)(s))
#define STREAM_FINI(s)          stream_fini((struct vsf_stream_t *)(s))
#define STREAM_WRITE(s, b)      stream_write((struct vsf_stream_t *)(s), (b))
#define STREAM_READ(s, b)       stream_read((struct vsf_stream_t *)(s), (b))
#define STREAM_GET_DATA_SIZE(s) stream_get_data_size((struct vsf_stream_t *)(s))
#define STREAM_GET_FREE_SIZE(s) stream_get_free_size((struct vsf_stream_t *)(s))
#define STREAM_GET_WBUF(s, p)   stream_get_wbuf((struct vsf_stream_t *)(s), (p))
#define STREAM_GET_RBUF(s, p)   stream_get_rbuf((struct vsf_stream_t *)(s), (p))
#define STREAM_CONNECT_RX(s)    stream_connect_rx((struct vsf_stream_t *)(s))
#define STREAM_CONNECT_TX(s)    stream_connect_tx((struct vsf_stream_t *)(s))
#define STREAM_DISCONNECT_RX(s) stream_disconnect_rx((struct vsf_stream_t *)(s))
#define STREAM_DISCONNECT_TX(s) stream_disconnect_tx((struct vsf_stream_t *)(s))

和標(biāo)準(zhǔn)接口只是大小寫的區(qū)別。