http.c

来自「mms client」· C语言 代码 · 共 2,402 行 · 第 1/5 页

static HTTPClient *client_create(int port, Connection *conn, Octstr *ip) {
    HTTPClient *p;

#ifdef HAVE_LIBSSL
    if (conn_get_ssl(conn))
        debug("gwlib.http", 0, "HTTP: Creating SSL-enabled HTTPClient for `%s', using cipher '%s'.",
              octstr_get_cstr(ip), SSL_get_cipher_version(conn_get_ssl(conn)));
    else
#endif    
        debug("gwlib.http", 0, "HTTP: Creating HTTPClient for `%s'.",
              octstr_get_cstr(ip));
    p = gw_malloc(sizeof(*p));
    p->port = port;
    p->conn = conn;
    p->ip = ip;
    p->state = reading_request_line;
    p->url = NULL;
    p->use_version_1_0 = 0;
    p->persistent_conn = 1;
    p->request = NULL;
    return p;
}


static void client_destroy(void *client) {
    HTTPClient *p;

    if (client == NULL)
        return;

    p = client;
    debug("gwlib.http", 0, "HTTP: Destroying HTTPClient area %p.", p);
    gw_assert_allocated(p, __FILE__, __LINE__, __func__);
    debug("gwlib.http", 0, "HTTP: Destroying HTTPClient for `%s'.",
          octstr_get_cstr(p->ip));
    conn_destroy(p->conn);
    octstr_destroy(p->ip);
    octstr_destroy(p->url);
    entity_destroy(p->request);
    gw_free(p);
}


static void client_reset(HTTPClient *p) {
    debug("gwlib.http", 0, "HTTP: Resetting HTTPClient for `%s'.",
          octstr_get_cstr(p->ip));
    p->state = reading_request_line;
    gw_assert(p->request == NULL);
}


/*
 * Checks whether the client connection is meant to be persistent or not.
 * Returns 1 for true, 0 for false.
 */

static int client_is_persistent(List *headers, int use_version_1_0) {
    Octstr *h = http_header_find_first(headers, "Connection");

    if (h == NULL) {
        return !use_version_1_0;
    } else {
        if (!use_version_1_0) {
            if (octstr_case_compare(h, octstr_imm("keep-alive")) == 0) {
                octstr_destroy(h);
                return 1;
            } else {
                octstr_destroy(h);
                return 0;
            }
        } else if (octstr_case_compare(h, octstr_imm("close")) == 0) {
            octstr_destroy(h);
            return 0;
        }
        octstr_destroy(h);
    }

    return 1;
}


/*
 * Port specific lists of clients with requests.
 */


struct port {
    List *clients_with_requests;
    Counter *active_consumers;
};


static Mutex *port_mutex = NULL;
static Dict *port_collection = NULL;


static void port_init(void) {
    port_mutex = mutex_create();
    port_collection = dict_create(1024, NULL);
}

static void port_shutdown(void) {
    mutex_destroy(port_mutex);
    dict_destroy(port_collection);
}


static Octstr *port_key(int port) {
    return octstr_format("%d", port);
}


static void port_add(int port) {
    Octstr *key;
    struct port *p;

    p = gw_malloc(sizeof(*p));
    p->clients_with_requests = list_create();
    list_add_producer(p->clients_with_requests);
    p->active_consumers = counter_create();

    key = port_key(port);
    mutex_lock(port_mutex);
    dict_put(port_collection, key, p);
    mutex_unlock(port_mutex);
    octstr_destroy(key);
}


static void port_remove(int port) {
    Octstr *key;
    struct port *p;

    key = port_key(port);
    mutex_lock(port_mutex);
    p = dict_remove(port_collection, key);
    mutex_unlock(port_mutex);
    octstr_destroy(key);

    list_remove_producer(p->clients_with_requests);
    while (counter_value(p->active_consumers) > 0)
        gwthread_sleep(0.1);    /* Reasonable use of busy waiting. */
    list_destroy(p->clients_with_requests, client_destroy);
    counter_destroy(p->active_consumers);
    gw_free(p);
}


static void port_put_request(HTTPClient *client) {
    Octstr *key;
    struct port *p;

    mutex_lock(port_mutex);
    key = port_key(client->port);
    p = dict_get(port_collection, key);
    gw_assert(p != NULL);
    list_produce(p->clients_with_requests, client);
    octstr_destroy(key);
    mutex_unlock(port_mutex);
}


static HTTPClient *port_get_request(int port) {
    Octstr *key;
    struct port *p;
    HTTPClient *client;

    mutex_lock(port_mutex);
    key = port_key(port);
    p = dict_get(port_collection, key);
    octstr_destroy(key);

    if (p == NULL) {
        client = NULL;
        mutex_unlock(port_mutex);
    } else {
        counter_increase(p->active_consumers);
        mutex_unlock(port_mutex);   /* Placement of this unlock is tricky. */
        client = list_consume(p->clients_with_requests);
        counter_decrease(p->active_consumers);
    }
    return client;
}


/*
 * Maximum number of servers (ports) we have open at the same time.
 */
enum {
    MAX_SERVERS = 32
};


/*
 * Variables related to server side implementation.
 */
static Mutex *server_thread_lock = NULL;
static volatile sig_atomic_t server_thread_is_running = 0;
static long server_thread_id = -1;
static FDSet *server_fdset = NULL;
static List *new_server_sockets = NULL;
static List *closed_server_sockets = NULL;
static int keep_servers_open = 0;


static int parse_request_line(int *method, Octstr **url,
                              int *use_version_1_0, Octstr *line) {
    List *words;
    Octstr *version;
    Octstr *method_str;
    int ret;

    words = octstr_split_words(line);
    if (list_len(words) != 3) {
        list_destroy(words, octstr_destroy_item);
        return -1;
    }

    method_str = list_get(words, 0);
    *url = list_get(words, 1);
    version = list_get(words, 2);
    list_destroy(words, NULL);

    if (octstr_compare(method_str, octstr_imm("GET")) == 0)
        *method = HTTP_METHOD_GET;
    else if (octstr_compare(method_str, octstr_imm("POST")) == 0)
        *method = HTTP_METHOD_POST;
    else if (octstr_compare(method_str, octstr_imm("HEAD")) == 0)
        *method = HTTP_METHOD_HEAD;
    else
        goto error;

    ret = parse_http_version(version);
    if (ret < 0)
        goto error;
    *use_version_1_0 = !ret;

    octstr_destroy(method_str);
    octstr_destroy(version);
    return 0;

    error:
    octstr_destroy(method_str);
    octstr_destroy(*url);
    octstr_destroy(version);
    *url = NULL;
    return -1;
}


static void receive_request(Connection *conn, void *data) {
    HTTPClient *client;
    Octstr *line;
    int ret;

    if (run_status != running) {
        conn_unregister(conn);
        return;
    }

    client = data;

    for (;;) {
        switch (client->state) {
            case reading_request_line:
                line = conn_read_line(conn);
                if (line == NULL) {
                    if (conn_eof(conn) || conn_read_error(conn))
                        goto error;
                    return;
                }
                ret = parse_request_line(&client->method, &client->url,
                                         &client->use_version_1_0, line);
                octstr_destroy(line);
                if (ret == -1)
                    goto error;
                /*
                 * RFC2616 (4.3) says we should read a message body if there
                 * is one, even on GET requests.
                 */
                client->request = entity_create(expect_body_if_indicated);
                client->state = reading_request;
                break;

            case reading_request:
                ret = entity_read(client->request, conn);
                if (ret < 0)
                    goto error;
                if (ret == 0) {
                    client->state = request_is_being_handled;
                    conn_unregister(conn);
                    port_put_request(client);
                }
                return;

            case sending_reply:
                if (conn_outbuf_len(conn) > 0)
                    return;
                /* Reply has been sent completely */
                if (!client->persistent_conn) {
                    client_destroy(client);
                    return;
                }
                /* Start reading another request */
                client_reset(client);
                break;

            default:
                panic(0, "Internal error: HTTPClient state is wrong.");
        }
    }

    error:
    client_destroy(client);
}


struct server {
    int fd;
    int port;
    int ssl;
};


static void server_thread(void *dummy) {
    struct pollfd tab[MAX_SERVERS];
    int ports[MAX_SERVERS];
    int ssl[MAX_SERVERS];
    long i, j, n, fd;
    int *portno;
    struct server *p;
    struct sockaddr_in addr;
    int addrlen;
    Connection *conn;
    HTTPClient *client;
    int ret;

    n = 0;
    while (run_status == running && keep_servers_open) {
        if (n == 0 || (n < MAX_SERVERS && list_len(new_server_sockets) > 0)) {
            p = list_consume(new_server_sockets);
            if (p == NULL) {
                debug("gwlib.http", 0, "HTTP: No new servers. Quitting.");
                break;
            }
            tab[n].fd = p->fd;
            tab[n].events = POLLIN;
            ports[n] = p->port;
            ssl[n] = p->ssl;
            ++n;
            gw_free(p);
        }

        if ((ret = gwthread_poll(tab, n, -1.0)) == -1) {
            if (errno != EINTR)
                warning(0, "HTTP: gwthread_poll failed.");
            continue;
        }

        for (i = 0; i < n; ++i) {
            if (tab[i].revents & POLLIN) {
                addrlen = sizeof(addr);
                fd = accept(tab[i].fd, (struct sockaddr *) &addr, &addrlen);
                if (fd == -1) {
                    error(errno, "HTTP: Error accepting a client.");
                    (void) close(tab[i].fd);
                    port_remove(ports[i]);
                    tab[i].fd = -1;
                    ports[i] = -1;
                    ssl[i] = 0;
                } else {
                    /*
                     * Be aware that conn_wrap_fd() will return NULL if SSL handshake
                     * has failed, so we only client_create() if there is an conn.
                     */             
                    if ((conn = conn_wrap_fd(fd, ssl[i]))) {
                        client = client_create(ports[i], conn, host_ip(addr));
                        conn_register(conn, server_fdset, receive_request, 
                                      client);
                    } else {
                        error(0, "HTTP: unsuccessfull SSL handshake for client `%s'",
                              octstr_get_cstr(host_ip(addr)));
                    }
                }
            }
        }

        while ((portno = list_extract_first(closed_server_sockets)) != NULL) {
            for (i = 0; i < n; ++i) {
                if (ports[i] == *portno) {
                    (void) close(tab[i].fd);
                    port_remove(ports[i]);
                    tab[i].fd = -1;
                    ports[i] = -1;
                    ssl[i] = 0;
                }
            }
            gw_free(portno);
        }

        j = 0;
        for (i = 0; i < n; ++i) {
            if (tab[i].fd != -1) {
                tab[j] = tab[i];
                ports[j] = ports[i];
                ssl[j] = ssl[i];
                ++j;
            }
        }
        n = j;
    }

    for (i = 0; i < n; ++i) {
        (void) close(tab[i].fd);
        port_remove(ports[i]);
    }
}


static void start_server_thread(void) {
    if (!server_thread_is_running) {
        /* 
         * To be really certain, we must repeat the test, but use the
         * lock first. If the test failed, however, we _know_ we've
         * already initialized. This strategy of double testing avoids
         * using the lock more than a few times at startup.
         */
        mutex_lock(server_thread_lock);
        if (!server_thread_is_running) {
            server_fdset = fdset_create();
            server_thread_id = gwthread_create(server_thread, NULL);
            server_thread_is_running = 1;
        }
        mutex_unlock(server_thread_lock);
    }
}


int http_open_port_if(int port, int ssl, Octstr *interface) {
    struct server *p;

    if (ssl) debug("gwlib.http", 0, "HTTP: Opening SSL server at port %d.", port);
    else debug("gwlib.http", 0, "HTTP: Opening server at port %d.", port);
    p = gw_malloc(sizeof(*p));
    p->port = port;
    p->ssl = ssl;
    p->fd = make_server_socket(port, (interface ? octstr_get_cstr(interface) : NULL));
    if (p->fd == -1) {
        gw_free(p);
        return -1;
    }

    port_add(port);
    list_produce(new_server_sockets, p);
    keep_servers_open = 1;
    start_server_thread();
    gwthread_wakeup(server_thread_id);

    return 0;
}


int http_open_port(int port, int ssl) {
    return http_open_port_if(port, ssl, NULL);
}


void http_close_port(int port) {
    int *p;

    p = gw_malloc(sizeof(*p));
    *p = port;
    list_produce(closed_server_sockets, p);
    gwthread_wakeup(server_thread_id);
}


void http_close_all_ports(void) {
    if (server_thread_id != -1) {
        keep_servers_open = 0;