zan-extension/src/factory/zanFactory.c
/**
 * [ReactorThread] dispatch request to worker
 */
[ReactorThread] dispatch request to worker
[ ReactorThread ] dispatch request to worker
static int zanFactory_dispatch(zanFactory *factory, swDispatchData *task)
{
    uint32_t schedule_key = 0;
    uint32_t send_len     = 0;
    uint16_t to_worker_id = -1;
    zanServer *serv = ServerG.serv;
    if (task->target_worker_id < 0)
    {
        schedule_key = task->data.info.fd;  
        to_worker_id = zanServer_worker_schedule(serv, task->data.info.networker_id, schedule_key);
    }
    else
    {
        to_worker_id = task->target_worker_id;
    }
    if (swEventData_is_stream(task->data.info.type))
    {
        swConnection *conn = zanServer_get_connection(serv, task->data.info.networker_id, task->data.info.fd);
        if (conn == NULL || conn->active == 0)
        {
            zanWarn("dispatch[type=%d] failed, connection#%d is not active.", task->data.info.type, task->data.info.fd);
            return ZAN_ERR;
        }
        if (conn->closed)
        {
            if (!(task->data.info.type == SW_EVENT_CLOSE && conn->close_force))
            {
                zanTrace("networker_id=%d, dispatch[type=%d] failed, connection#%d:[session_id=%d] is closed by server.",
                         task->data.info.networker_id, task->data.info.type, task->data.info.fd, conn->session_id);
                return ZAN_OK;
            }
            else
            {
                zanWarn("error: networker_id=%d, task_type=%d, connection_fd=%d, session_id=%d",
                         task->data.info.networker_id, task->data.info.type, task->data.info.fd, conn->session_id);
                return ZAN_ERR;
            }
        }
        task->data.info.fd = conn->session_id;
        task->data.info.from_fd = conn->from_fd;
        zanDebug("send2worker: connection_fd=%d, session_id=%d, from_fd=%d, len=%d, worker_id=%d",
                  task->data.info.fd, conn->session_id, conn->from_fd, send_len, to_worker_id);
    }
    send_len = sizeof(task->data.info) + task->data.info.len;
    return zanNetworker_send2worker((void *) &(task->data), send_len, to_worker_id);
}
static int zanFactory_dispatch ( zanFactory * factory , swDispatchData * task ) { uint32_t schedule_key = 0 ; uint32_t send_len = 0 ; uint16_t to_worker_id = -1 ; zanServer * serv = ServerG . serv ; if ( task -> target_worker_id < 0 ) { schedule_key = task -> data . info . fd ; to_worker_id = zanServer_worker_schedule ( serv , task -> data . info . networker_id , schedule_key ) ; } else { to_worker_id = task -> target_worker_id ; } if ( swEventData_is_stream ( task -> data . info . type ) ) { swConnection * conn = zanServer_get_connection ( serv , task -> data . info . networker_id , task -> data . info . fd ) ; if ( conn == NULL || conn -> active == 0 ) { zanWarn ( " " , task -> data . info . type , task -> data . info . fd ) ; return ZAN_ERR ; } if ( conn -> closed ) { if ( ! ( task -> data . info . type == SW_EVENT_CLOSE && conn -> close_force ) ) { zanTrace ( " " , task -> data . info . networker_id , task -> data . info . type , task -> data . info . fd , conn -> session_id ) ; return ZAN_OK ; } else { zanWarn ( " " , task -> data . info . networker_id , task -> data . info . type , task -> data . info . fd , conn -> session_id ) ; return ZAN_ERR ; } } task -> data . info . fd = conn -> session_id ; task -> data . info . from_fd = conn -> from_fd ; zanDebug ( " " , task -> data . info . fd , conn -> session_id , conn -> from_fd , send_len , to_worker_id ) ; } send_len = sizeof ( task -> data . info ) + task -> data . info . len ; return zanNetworker_send2worker ( ( void * ) & ( task -> data ) , send_len , to_worker_id ) ; }
[ReactorThread] dispatch request to worker
[ ReactorThread ] dispatch request to worker
//1. get target_worker_id //accept fd //2. send data to worker //conn active close, discard data. //converted fd to session_id