/*
  +----------------------------------------------------------------------+
  | Zan                                                                  |
  +----------------------------------------------------------------------+
  | Copyright (c) 2016-2017 Zan Group <https://github.com/youzan/zan>    |
  +----------------------------------------------------------------------+
  | This source file is subject to version 2.0 of the Apache license,    |
  | that is bundled with this package in the file LICENSE, and is        |
  | available through the world-wide-web at the following url:           |
  | http://www.apache.org/licenses/LICENSE-2.0.html                      |
  | If you did not receive a copy of the Apache2.0 license and are unable|
  | to obtain it through the world-wide-web, please send a note to       |
  | zan@zanphp.io so we can mail you a copy immediately.                 |
  +----------------------------------------------------------------------+
  |         Zan Group   <zan@zanphp.io>                                  |
  +----------------------------------------------------------------------+
*/

#include "zanGlobalDef.h"
#include "zanServer.h"
#include "zanWorkers.h"
#include "zanFactory.h"
#include "zanLog.h"

typedef struct _zanNotify_data
{
    long target_worker_id;
    swDataHead _send;
} zanNotify_data;

static int zanFactory_start(zanFactory *factory);
static int zanFactory_notify(zanFactory *factory, swDataHead *event);
static int zanFactory_dispatch(zanFactory *factory, swDispatchData *buf);
static int zanFactory_finish(zanFactory *factory, swSendData *data);
static int zanFactory_shutdown(zanFactory *factory);
static int zanFactory_end(zanFactory *factory, int session_id);

int zanFactory_create(zanFactory *factory)
{
    if (!factory)
    {
        zanError("error, factory is null.");
        return ZAN_ERR;
    }

    factory->start    = zanFactory_start;
    factory->end      = zanFactory_end;
    factory->notify   = zanFactory_notify;
    factory->finish   = zanFactory_finish;
    factory->shutdown = zanFactory_shutdown;
    factory->dispatch = zanFactory_dispatch;
    return ZAN_OK;
}

static int zanFactory_shutdown(zanFactory *factory)
{
    if (!factory)
    {
        zanError("factory is null");
        return ZAN_ERR;
    }

    zanDebug("factory shutdown.");
    return ZAN_OK;
}

static int zanFactory_start(zanFactory *factory)
{
    if (!factory)
    {
        zanError("factory is null");
        return ZAN_ERR;
    }

    if (zan_start_worker_processes() < 0)
    {
        zanError("zan_start_worker_processes failed.");
        return ZAN_ERR;
    }

    ////TODO
    ////factory->finish = swFactory_finish;

    return ZAN_OK;
}

//close connection, and notify to worker.
//swReactorThread_onClose
static int zanFactory_notify(zanFactory *factory, swDataHead *ev)
{
    zanNotify_data notify_data;
    if (!factory || !ev)
    {
        zanError("factory=%p or ev=%p is null", factory, ev);
        return ZAN_ERR;
    }

    memcpy(&notify_data._send, ev, sizeof(swDataHead));
    notify_data._send.len = 0;
    notify_data.target_worker_id = -1;
    return factory->dispatch(factory, (swDispatchData *) &notify_data);
}

/**
 * [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;

    //1. get target_worker_id
    if (task->target_worker_id < 0)
    {
        schedule_key = task->data.info.fd;  //accept fd
        to_worker_id = zanServer_worker_schedule(serv, task->data.info.networker_id, schedule_key);
    }
    else
    {
        to_worker_id = task->target_worker_id;
    }

    //2. send data to worker
    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;
        }
        //conn active close, discard data.
        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;
            }
        }

        //converted fd to session_id
        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);
}

//send data to client
static int zanFactory_finish(zanFactory *factory, swSendData *resp)
{
    if (!factory || !resp)
    {
        zanError("factory=%p or resp=%p is null", factory, resp);
        return ZAN_ERR;
    }

    zanServer *serv = (zanServer *)ServerG.serv;

    int session_id = resp->info.fd;
    swConnection *conn = zanServer_verify_connection(serv, session_id);
    if (!conn)
    {
        zanWarn("session#fd=%d does not exist.", session_id);
        return ZAN_ERR;
    }
    else if ((conn->closed || conn->removed) && resp->info.type != SW_EVENT_CLOSE)
    {
        int _len = resp->length > 0 ? resp->length : resp->info.len;
        zanWarn("send %d byte failed, because session#fd=%d is closed, length=%d.", _len, session_id, resp->length);
        return ZAN_ERR;
    }
    else if (conn->overflow)
    {
        zanWarn("send failed, session#fd=%d output buffer has been overflowed.", session_id);
        return ZAN_ERR;
    }

    swEventData ev_data;
    memset(&ev_data, 0, sizeof(ev_data));
    ev_data.info.fd   = session_id;
    ev_data.info.type = resp->info.type;

    zanWorker *worker  = zanServer_get_worker(serv, ServerWG.worker_id);

    //Big response, use shared memory
    if (resp->length > 0)
    {
        if (worker->send_shm == NULL)
        {
            zanWarn("send failed, data is too big.");
            return ZAN_ERR;
        }

        swPackage_response response;

        worker->lock.lock(&worker->lock);
        response.length = resp->length;
        response.worker_id = ServerWG.worker_id;

        zanDebug("BigPackage, length=%d|worker_id=%d", response.length, response.worker_id);

        ev_data.info.from_fd = ZAN_RESPONSE_BIG;
        ev_data.info.len = sizeof(response);

        memcpy(ev_data.data, &response, sizeof(response));
        memcpy(worker->send_shm, resp->data, resp->length);
    }
    else
    {
        //copy data
        memcpy(ev_data.data, resp->data, resp->info.len);
        ev_data.info.len = resp->info.len;
        ev_data.info.from_fd = ZAN_RESPONSE_SMALL;
    }

    int sendn = ev_data.info.len + sizeof(resp->info);

    //ev_data.info.from_id = conn->from_id;
    ev_data.info.from_id = -1; ////////////////TODO:::no
    zanTrace("[Worker] send: worker_id=%d, session_id=%d, sendn=%d|type=%d|content=%s",
             ServerWG.worker_id, session_id, sendn, resp->info.type, resp->data);

    int ret = zanWorker_send2networker(&ev_data, sendn, session_id);
    if (ret < 0)
    {
        zanError("sendto to reactor failed.");
    }

    return ret;
}

//关闭连接
//TODO:::
//1. server: close 接口
//2. SW_EVENT_CLOSE 事件
static int zanFactory_end(zanFactory *factory, int session_id)
{
    zanServer *serv = (zanServer *)ServerG.serv;
    swSendData _send;
    swDataHead info;

    bzero(&_send, sizeof(_send));
    _send.info.fd   = session_id;
    _send.info.len  = 0;
    _send.info.type = SW_EVENT_CLOSE;

    //1. get and verify connection, then close the conn
    swConnection *conn = zanServer_get_connection_by_sessionId(serv, session_id);
    if (conn == NULL || conn->active == 0)
    {
        zanWarn("can not close. Connection[%d] not found.", _send.info.fd);
        return ZAN_ERR;
    }
    else if (conn->close_force)
    {
		conn->closing = 1;
        if (serv->onClose != NULL)
        {
            info.fd = session_id;
            info.from_id =  conn->from_id;
            info.from_fd =  conn->from_fd;
            info.networker_id = conn->networker_id;
            serv->onClose(serv, &info);
        }
        conn->closing = 0;
        conn->closed = 1;
        return factory->finish(factory, &_send);
    }
    else if (conn->closing)
    {
        zanWarn("The connection[fd=%d] is closing.", session_id);
        return ZAN_ERR;
    }
    else if (conn->closed)
    {
        zanWarn("The connection[fd=%d] is closed.", session_id);
        return ZAN_ERR;
    }
    else
    {
        conn->closing = 1;
        if (serv->onClose != NULL)
        {
            info.fd = session_id;
            info.from_id =  conn->from_id;
            info.from_fd =  conn->from_fd;
            info.networker_id = conn->networker_id;
            serv->onClose(serv, &info);
        }
        conn->closing = 0;
        conn->closed = 1;
        return factory->finish(factory, &_send);
    }
}
