ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: opt / alt / php70 / usr / include / php / ext / swoole / include

#pragma once

#include "swoole.h"
#include "coroutine.h"

#include <sys/stat.h>

#include <iostream>
#include <string>
#include <list>
#include <queue>

namespace swoole
{
class Channel
{
public:
    enum opcode
    {
        PRODUCER = 1,
        CONSUMER = 2,
    };

struct timer_msg_t
    {
        Channel *chan;
        enum opcode type;
        Coroutine *co;
        bool error;
        swTimer_node *timer;
    };

void* pop(double timeout = -1);
    bool push(void *data, double timeout = -1);
    bool close();

Channel(size_t _capacity = 1) :
            capacity(_capacity)
    {
    }

~Channel()
    {
        SW_ASSERT(producer_queue.empty() && consumer_queue.empty());
    }

inline bool is_closed()
    {
        return closed;
    }

inline bool is_empty()
    {
        return data_queue.size() == 0;
    }

inline bool is_full()
    {
        return data_queue.size() == capacity;
    }

inline size_t length()
    {
        return data_queue.size();
    }

inline size_t consumer_num()
    {
        return consumer_queue.size();
    }

inline size_t producer_num()
    {
        return producer_queue.size();
    }

inline void* pop_data()
    {
        if (data_queue.size() == 0)
        {
            return nullptr;
        }
        void *data = data_queue.front();
        data_queue.pop();
        return data;
    }

protected:
    size_t capacity = 1;
    bool closed = false;
    std::list<Coroutine *> producer_queue;
    std::list<Coroutine *> consumer_queue;
    std::queue<void *> data_queue;

static void timer_callback(swTimer *timer, swTimer_node *tnode);

void yield(enum opcode type);

inline void consumer_remove(Coroutine *co)
    {
        consumer_queue.remove(co);
    }

inline void producer_remove(Coroutine *co)
    {
        producer_queue.remove(co);
    }

inline Coroutine* pop_coroutine(enum opcode type)
    {
        Coroutine* co;
        if (type == PRODUCER)
        {
            co = producer_queue.front();
            producer_queue.pop_front();
            swTraceLog(SW_TRACE_CHANNEL, "resume producer cid=%ld", co->get_cid());
        }
        else // if (type == CONSUMER)
        {
            co = consumer_queue.front();
            consumer_queue.pop_front();
            swTraceLog(SW_TRACE_CHANNEL, "resume consumer cid=%ld", co->get_cid());
        }
        return co;
    }
};
};

[-] error.h [edit]
[-] http2.h [edit]
[-] mime_types.h [edit]
[-] coroutine_c_api.h [edit]
[-] context.h [edit]
[-] redis.h [edit]
[-] table.h [edit]
[-] uthash.h [edit]
[-] socket.h [edit]
[-] hashmap.h [edit]
[-] hash.h [edit]
[-] list.h [edit]
[-] socket_hook.h [edit]
[-] connection.h [edit]
[-] lru_cache.h [edit]
[-] buffer.h [edit]
[-] asm_context.h [edit]
[-] atomic.h [edit]
[-] mqtt.h [edit]
[-] sha1.h [edit]
[-] ring_queue.h [edit]
[-] coroutine.h [edit]
[-] api.h [edit]
[-] channel.h [edit]
[-] rbtree.h [edit]
[-] win.h [edit]
[-] websocket.h [edit]
[-] client.h [edit]
[-] unix.h [edit]
[-] async.h [edit]
[-] file_hook.h [edit]
[-] swoole.h [edit]
[-] array.h [edit]
[-] http.h [edit]
[-] base64.h [edit]
[+] ..
[-] socks5.h [edit]
[-] server.h [edit]
[-] heap.h [edit]