学步园

enum {
    SPICE_CHANNEL_MAIN = 1,
    SPICE_CHANNEL_DISPLAY,
    SPICE_CHANNEL_INPUTS,
    SPICE_CHANNEL_CURSOR,
    SPICE_CHANNEL_PLAYBACK,
    SPICE_CHANNEL_RECORD,
    SPICE_CHANNEL_TUNNEL, // 没定义USE_TUNNEL，则此Channel无效

    SPICE_END_CHANNEL
};

void RedClient::create_channel(uint32_t type, uint32_t id)

{

    ChannelFactory* factory = find_factory(type);

    RedChannel* channel = factory->construct(*this, id);

    _channels.push_back(channel);

 

    channel->start();   // 启动RedChannel类的工作线程

    channel->connect(); // 激活condition

 

    _migrate.add_channel(new MigChannel(type, id, channel->get_common_caps(), channel->get_caps()));

}

typedef struct Channel {

    struct Channel *next;

    uint32_t type;

    uint32_t id;

    int num_common_caps;

    uint32_t *common_caps;

    int num_caps;

    uint32_t *caps;

    void (*link)(struct Channel *, RedsStreamContext *peer, int migration, int num_common_caps,

                 uint32_t *common_caps, int num_caps, uint32_t *caps);

    void (*shutdown)(struct Channel *);

    void (*migrate)(struct Channel *);

    void *data;

} Channel;

reds_channel = spice_new0(Channel, 1);

    reds_channel->type = SPICE_CHANNEL_DISPLAY;

    reds_channel->id = qxl->id;

    reds_channel->link = red_dispatcher_set_peer;

    reds_channel->shutdown = red_dispatcher_shutdown_peer;

    reds_channel->migrate = red_dispatcher_migrate;

    reds_channel->data = dispatcher;

reds_register_channel(reds_channel);

   typedef struct AsyncRead {

    RedsStreamContext *peer; // 用来处理网络数据流的数据对象

    void *opaque;  // 传递数据用的，通常作为done函数的参数

    uint8_t *now;  // 数据起始，用来保存接收数据

    uint8_t *end;  // 数据结束，通常用来计算数据长度

    void (*done)(void *opaque); // 消息处理函数

    void (*error)(void *opaque, int err); // 错误处理函数

   } AsyncRead;

 static void async_read_handler(int fd, int event, void *data)

{

    AsyncRead *obj = (AsyncRead *)data;

 

    for (;;) {

        int n = obj->end - obj->now;

        if ((n = obj->peer->cb_read(obj->peer->ctx, obj->now, n)) <= 0) {

           // 异常处理省略……

        } else { // 正常情况处理

            obj->now += n;

            if (obj->now == obj->end) { // 接收数据完毕了就执行done，否则继续循环接收数据

                async_read_clear_handlers(obj);

                obj->done(obj->opaque);

                return;

            }

        }

    }

}

SPICE_CHANNEL_DISPLAY   red_dispatcher_set_peer

SPICE_CHANNEL_CURSOR        red_dispatcher_set_cursor_peer

SPICE_CHANNEL_INPUTS        inputs_link

SPICE_CHANNEL_PLAYBACK snd_set_playback_peer

SPICE_CHANNEL_RECORD        snd_set_record_peer

static void handle_new_display_channel(RedWorker *worker, RedsStreamContext *peer, int migrate)

{

    DisplayChannel *display_channel;

    size_t stream_buf_size;

 

     // 先断开原来的Channel连接，所做工作基本与此函数后面的工作相反

    red_disconnect_display((RedChannel *)worker->display_channel);

 

    // __new_channel函数会new一个RedChannel，并用该函数的参数初始化RedChannel的成员

    // RedChannel数据成员比较多，只拣最主要的说明一下：

// struct RedChannel {

//     EventListener listener;     // 事件触发器，网络事件处理用

//     spice_parse_channel_func_t parser; // 用途待查，网络通信解码器？？

//     struct RedWorker *worker; // 用以保存RedWorker指针

//     RedsStreamContext *peer;  // 用以保存RedsStreamContext指针，此指针在响应客户端连接时创建，网络通信的主要处理对象

//

//     Ring pipe; // 用途待查明，声明为Ring，表明是一个链表，而非一个链表内的一项。如果声明为RingItem，表明其容器为一个Item

//

//     disconnect_channel_proc disconnect;  // 下面四个是回调函数

//     hold_item_proc hold_item;       // 针对不同的Channel有不同的实现

//     release_item_proc release_item;    // 通过__new_channel函数

//     handle_message_proc handle_message; // 的参数传入来初始化

// };

// 初始化上述结构后，__new_channel调用

// epoll_ctl(worker->epoll, EPOLL_CTL_ADD, peer->socket, &event)

// 将该channel对应的网络连接socket加入到red_worker_main的epoll事件监听// 列表中，以在red_worker_main中处理此Channel的网络事件

if (!(display_channel = (DisplayChannel *)__new_channel(worker,

 sizeof(*display_channel),SPICE_CHANNEL_DISPLAY, peer,migrate,

 handle_channel_events,

red_disconnect_display,

display_channel_hold_item,

display_channel_release_item,

display_channel_handle_message))) {

        return;

    }

 

    ring_init(&display_channel->palette_cache_lru);

    red_display_init_streams(display_channel);

    red_display_share_stream_buf(display_channel);

    red_display_init_glz_data(display_channel);

    worker->display_channel = display_channel;

 

    on_new_display_channel(worker);

}

static void red_connect_cursor(RedWorker *worker, RedsStreamContext *peer, int migrate)

{

    CursorChannel *channel;

    red_disconnect_cursor((RedChannel *)worker->cursor_channel);

if (!(channel = (CursorChannel *)__new_channel(worker, sizeof(*channel),

           SPICE_CHANNEL_CURSOR, peer, migrate,

handle_channel_events,

red_disconnect_cursor,

cursor_channel_hold_item,

cursor_channel_release_item,

channel_handle_message))) {

        return;

    }

 

    ring_init(&channel->cursor_cache_lru);

    worker->cursor_channel = channel;

    on_new_cursor_channel(worker);

}

static void inputs_link(Channel *channel, RedsStreamContext *peer,

int migration,int num_common_caps, uint32_t *common_caps, int num_caps,

uint32_t *caps)

{

    InputsState *inputs_state;

    inputs_state = spice_new0(InputsState, 1);

 

    // 一些初始化……

 

    peer->watch = core->watch_add(peer->socket, SPICE_WATCH_EVENT_READ,

                                  inputs_event, inputs_state);

 

    SpiceMarshaller *m;

    SpiceMsgInputsInit inputs_init;

    m = marshaller_new_for_outgoing(inputs_state, SPICE_MSG_INPUTS_INIT);

    inputs_init.keyboard_modifiers = kbd_get_leds(keyboard);

    spice_marshall_msg_inputs_init(m, &inputs_init);

}

static void snd_set_record_peer(Channel *channel, RedsStreamContext *peer,

int migration, int num_common_caps, uint32_t *common_caps, int num_caps,

                                uint32_t *caps)

{

    snd_disconnect_channel(worker->connection);

if (!(record_channel = (RecordChannel *)__new_channel(worker,

 sizeof(*record_channel),SPICE_CHANNEL_RECORD,peer,migration,

        snd_record_send,

snd_record_handle_message,

snd_record_on_message_done,

snd_record_cleanup))) {

        goto error_2;

    }

 

    on_new_record_channel(worker);

 

    if (worker->active) {

        spice_server_record_start(st->sin);

    }

    snd_record_send(worker->connection);

    return;

}

static void inputs_handle_input(void *opaque, size_t size, uint32_t type, void *message)

{

    InputsState *state = (InputsState *)opaque;

    uint8_t *buf = (uint8_t *)message;

    SpiceMarshaller *m;

 

    switch (type) {

    case SPICE_MSGC_INPUTS_KEY_DOWN:    

    case SPICE_MSGC_INPUTS_KEY_UP:

    case SPICE_MSGC_INPUTS_MOUSE_MOTION:

    case SPICE_MSGC_INPUTS_MOUSE_POSITION:

    case SPICE_MSGC_INPUTS_MOUSE_PRESS:

    case SPICE_MSGC_INPUTS_MOUSE_RELEASE:

    case SPICE_MSGC_INPUTS_KEY_MODIFIERS:

    case SPICE_MSGC_DISCONNECTING:

        break;

    default:

        red_printf("unexpected type %d", type);

    }

}

static int line_out_run (HWVoiceOut *hw, int live)

{

    SpiceVoiceOut *out = container_of (hw, SpiceVoiceOut, hw);

    decr = rate_get_samples (&hw->info, &out->rate);

 

    while (samples) {

        if (!out->frame) {

            spice_server_playback_get_buffer (&out->sin, &out->frame, &out->fsize); // 从spice中获取缓冲区

            out->fpos = out->frame;

        }

        if (out->frame) {

            hw->clip (out->fpos, hw->mix_buf + rpos, len); // 虚拟声卡捕获音频数据应该是放在mix_buf中

            out->fsize -= len;

            out->fpos  += len;

            if (out->fsize == 0) {

                spice_server_playback_put_samples (&out->sin, out->frame);

// 往缓冲区写数据，实际会将数据发送到客户端

                out->frame = out->fpos = NULL;

            }

        }

        rpos = (rpos + len) % hw->samples;

        samples -= len;

    }

}

 void *red_worker_main(void *arg)

{

    RedWorker worker;

 

    red_init(&worker, (WorkerInitData *)arg);

    red_init_quic(&worker);

    red_init_lz(&worker);

    red_init_jpeg(&worker);

    red_init_zlib(&worker);

    worker.epoll_timeout = INF_EPOLL_WAIT;

    for (;;) {

        num_events = epoll_wait(worker.epoll, events, MAX_EPOLL_SOURCES, worker.epoll_timeout);

        red_handle_streams_timout(&worker);

 

        if (worker.display_channel && worker.display_channel->glz_dict) {

              red_display_handle_glz_drawables_to_free(worker.display_channel);

        }

 

for (event = events, end = event + num_events; event < end; event++) {

            EventListener *evt_listener = (EventListener *)event->data.ptr;

 

            if (evt_listener->refs > 1) {

                evt_listener->action(evt_listener, event->events);

                if (--evt_listener->refs) {

                    continue;

                }

            }

            free(evt_listener); // refs == 0 , release it!

        }

 

        if (worker.running) {

            int ring_is_empty;

// 处理Cursor command，从虚拟设备中获取数据

            red_process_cursor(&worker, MAX_PIPE_SIZE, &ring_is_empty);

// 处理Display command，从虚拟设备中获取实时数据

            red_process_commands(&worker, MAX_PIPE_SIZE,

&ring_is_empty);

}

        red_push(&worker); // 往客户端push显示命令

    }

    red_printf("exit");

    return 0;

}

class RedChannel::MessageHandler {

public:

    MessageHandler() {}

    virtual ~MessageHandler() {}

    virtual void handle_message(RedPeer::CompoundInMessage& message) = 0;

};

template <class HandlerClass, unsigned int channel_id>

class MessageHandlerImp: public RedChannel::MessageHandler {

public:

    MessageHandlerImp(HandlerClass& obj);

    ~MessageHandlerImp() { delete [] _handlers; };

    virtual void handle_message(RedPeer::CompoundInMessage& message);

    typedef void (HandlerClass::*Handler)(RedPeer::InMessage* message);

    void set_handler(unsigned int id, Handler handler);

 

private:

    HandlerClass& _obj;

    unsigned int _max_messages;

    spice_parse_channel_func_t _parser;

    Handler *_handlers;

};

class MainChannelLoop: public MessageHandlerImp<RedClient, SPICE_CHANNEL_MAIN> {

public:

    MainChannelLoop(RedClient& client): MessageHandlerImp<RedClient, SPICE_CHANNEL_MAIN>(client) {}

};

RedChannel::worker_main(){

        RedChannel::run(){

       // 这里用了condition 相关的pthread库函数来进行信号控制

// pthread_cond_signal、pthread_cond_wait……

       // 工作线程启动后会等待RedClient的connect、disconnect、quit等操作

       // RedClient的connect会pthread_cond_signal(cond)

       _action_cond.wait --》 pthread_cond_wait(cond) // 等吧

 

       // _action 用来标识等来的是个什么操作，一共有三个：

// CONNECT_ACTION、DISCONNECT_ACTION、QUIT_ACTION

       // 后面两个是做一些清理工作，主要是第一个：CONNECT_ACTION

       case _action of CONNECT_ACTION:{

          RedChannelBase::connect // 调用基类的connect，进行网络连接

          _marshallers = spice_message_marshallers_get // 初始化marshallers

          on_connect --> _migrate.add_channel()

          on_event {send_messages,recive_messages}

       

          // ProcessLoop 消息循环

          _loop.add_socket(*this); // --> add_event, 往_loop中的

// _event_sources增加event，以进行监听……

// 注意了：上面使用了EventSources中的socket类，用于网络事件的监听

// 具体方法参见代码：

// HANDLE event = CreateEvent(NULL, FALSE, FALSE, NULL); // 创建一个Event，然后：

// WSAEventSelect(socket.get_socket(), event,FD_READ | FD_WRITE | FD_CLOSE)

// 上面这个函数将event 与 socket绑定，当socket上有事件被激活时，event也将被激活

 

       _loop.run() {

           _event_sources.wait_events(){

            DWORD wait_res = MsgWaitForMultipleObjectsEx(_handles.size(),  &_handles[0], timeout_ms,                                                 QS_ALLINPUT, 0); // 此函数用来等待事件，对于同步事件，此函数返回前会修改事件状态

                       

            int event_index = wait_res - WAIT_OBJECT_0;

            _events[event_index]->action() --> RedChannel::on_event{

               send_messages(){

                get_outgoing_message

                send

               }

               recive_messages(){

                RedPeer::recive()

                on_message_recived()

                _message_handler->handle_message(*(*message)){

                   (_obj.*_handlers[type])(&main_message); // 通过实例调用其成员函数，与SetHandler是所赋的成员函数对象一致

                }

               }

            }

           }   

void Application::register_channels()

{

    if (_enabled_channels[SPICE_CHANNEL_DISPLAY]) { //注册channelFactory

        _client.register_channel_factory(DisplayChannel::Factory());

    }

    ……

}

void RedClient::handle_channels(RedPeer::InMessage* message)

{

    SpiceMsgChannels *init = (SpiceMsgChannels *)message->data();

    SpiceChannelId* channels = init->channels;

    for (unsigned int i = 0; i < init->num_of_channels; i++) {

        create_channel(channels[i].type, channels[i].id);

    }

}

void RedClient::create_channel(uint32_t type, uint32_t id)

{

    ChannelFactory* factory = find_factory(type);

    RedChannel* channel = factory->construct(*this, id);

    _channels.push_back(channel);

    channel->start();   // 创建工作线程

    channel->connect(); // 工作线程会等待CONNECT_ACTION condition来激活线程

_migrate.add_channel(new MigChannel(type, id, channel->get_common_caps(), channel->get_caps()));

}