jk_ajp_common.c

jboss与apache集成的中间件,详情请参看文档说明.

     */
    if (ajp_connection_tcp_get_message(ae, msg, l) != JK_TRUE) {
        jk_log(l, JK_LOG_INFO,
               "awaited reply cpong, not received");
        JK_TRACE_EXIT(l);
        return JK_FALSE;
    }

    if ((cmd = jk_b_get_byte(msg)) != AJP13_CPONG_REPLY) {
        jk_log(l, JK_LOG_INFO,
               "awaited reply cpong, received %d instead",
               cmd);
        JK_TRACE_EXIT(l);
        return JK_FALSE;
    }

    JK_TRACE_EXIT(l);
    return JK_TRUE;
}

int ajp_connect_to_endpoint(ajp_endpoint_t * ae, jk_logger_t *l)
{
    char buf[32];
    int rc = JK_TRUE;

    JK_TRACE_ENTER(l);

    ae->sd = jk_open_socket(&ae->worker->worker_inet_addr,
                            ae->worker->keepalive,
                            ae->worker->socket_timeout,
                            ae->worker->socket_buf, l);
    if (ae->sd >= 0) {
        if (JK_IS_DEBUG_LEVEL(l)) {
            jk_log(l, JK_LOG_DEBUG,
                   "Connected socket %d to (%s)",
                   ae->sd,
                   jk_dump_hinfo(&ae->worker->worker_inet_addr, buf));
        }
        /* set last_access only if needed */
        if (ae->worker->cache_timeout > 0 || ae->worker->recycle_timeout > 0)
            ae->last_access = time(NULL);
        /* Check if we must execute a logon after the physical connect */
        if (ae->worker->logon != NULL) {
            rc = ae->worker->logon(ae, l);
            JK_TRACE_EXIT(l);
            return rc;
        }
        /* should we send a CPING to validate connection ? */
        if (ae->worker->connect_timeout > 0) {
            rc = ajp_handle_cping_cpong (ae,
                        ae->worker->connect_timeout, l);
            JK_TRACE_EXIT(l);
            return rc;
        }
        JK_TRACE_EXIT(l);
        return JK_TRUE;
    }

    jk_log(l, JK_LOG_INFO,
           "Failed opening socket to (%s) with (errno=%d)",
           jk_dump_hinfo(&ae->worker->worker_inet_addr, buf), errno);
    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

/*
 * Send a message to endpoint, using corresponding PROTO HEADER
 */

int ajp_connection_tcp_send_message(ajp_endpoint_t * ae,
                                    jk_msg_buf_t *msg, jk_logger_t *l)
{
    int rc;

    JK_TRACE_ENTER(l);
    if (ae->proto == AJP13_PROTO) {
        jk_b_end(msg, AJP13_WS_HEADER);
        if (JK_IS_DEBUG_LEVEL(l))
            jk_dump_buff(l, JK_LOG_DEBUG, "sending to ajp13", msg);
    }
    else if (ae->proto == AJP14_PROTO) {
        jk_b_end(msg, AJP14_WS_HEADER);
        if (JK_IS_DEBUG_LEVEL(l))
            jk_dump_buff(l, JK_LOG_DEBUG, "sending to ajp14", msg);
    }
    else {
        jk_log(l, JK_LOG_ERROR,
               "unknown protocol %d, supported are AJP13/AJP14", ae->proto);
        JK_TRACE_EXIT(l);
        return JK_FATAL_ERROR;
    }

    if ((rc = jk_tcp_socket_sendfull(ae->sd, msg->buf,
                                     msg->len)) > 0) {
        ae->endpoint.wr += msg->len;
        JK_TRACE_EXIT(l);
        return JK_TRUE;
    }
    jk_log(l, JK_LOG_ERROR,
           "sendfull returned %d with errno=%d ", rc, errno);

    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

/*
 * Receive a message from endpoint, checking PROTO HEADER
 */

int ajp_connection_tcp_get_message(ajp_endpoint_t * ae,
                                   jk_msg_buf_t *msg, jk_logger_t *l)
{
    unsigned char head[AJP_HEADER_LEN];
    int rc;
    int msglen;
    unsigned int header;
    char buf[32];

    JK_TRACE_ENTER(l);

    rc = jk_tcp_socket_recvfull(ae->sd, head, AJP_HEADER_LEN);

    if (rc < 0) {
        if (rc == JK_SOCKET_EOF) {
            jk_log(l, JK_LOG_INFO,
                   "Tomcat has forced a connection close for socket %d",
                   ae->sd);
            JK_TRACE_EXIT(l);
        }
        else {
            jk_log(l, JK_LOG_ERROR,
                   "Can't receive the response message from tomcat, "
                   "network problems or tomcat is down (%s), err=%d",
                   jk_dump_hinfo(&ae->worker->worker_inet_addr, buf), rc);
             JK_TRACE_EXIT(l);
        }
        return JK_FALSE;
    }
    ae->endpoint.rd += rc;
    header = ((unsigned int)head[0] << 8) | head[1];

    if (ae->proto == AJP13_PROTO) {
        if (header != AJP13_SW_HEADER) {

            if (header == AJP14_SW_HEADER) {
                jk_log(l, JK_LOG_ERROR,
                       "received AJP14 reply on an AJP13 connection from %s",
                       jk_dump_hinfo(&ae->worker->worker_inet_addr, buf));
            }
            else {
                jk_log(l, JK_LOG_ERROR,
                       "wrong message format 0x%04x from %s",
                       header, jk_dump_hinfo(&ae->worker->worker_inet_addr,
                                             buf));
            }
            JK_TRACE_EXIT(l);
            return JK_FALSE;
        }
    }
    else if (ae->proto == AJP14_PROTO) {
        if (header != AJP14_SW_HEADER) {

            if (header == AJP13_SW_HEADER) {
                jk_log(l, JK_LOG_ERROR,
                       "received AJP13 reply on an AJP14 connection from %s",
                       jk_dump_hinfo(&ae->worker->worker_inet_addr, buf));
            }
            else {
                jk_log(l, JK_LOG_ERROR,
                       "wrong message format 0x%04x from %s",
                       header, jk_dump_hinfo(&ae->worker->worker_inet_addr,
                                             buf));
            }
            JK_TRACE_EXIT(l);
            return JK_FALSE;
        }
    }

    msglen = ((head[2] & 0xff) << 8);
    msglen += (head[3] & 0xFF);

    if (msglen > msg->maxlen) {
        jk_log(l, JK_LOG_ERROR,
               "wrong message size %d %d from %s",
               msglen, msg->maxlen,
               jk_dump_hinfo(&ae->worker->worker_inet_addr, buf));
        JK_TRACE_EXIT(l);
        return JK_FALSE;
    }

    msg->len = msglen;
    msg->pos = 0;

    rc = jk_tcp_socket_recvfull(ae->sd, msg->buf, msglen);
    if (rc < 0) {
        jk_log(l, JK_LOG_ERROR,
               "ERROR: can't receive the response message from tomcat, "
               "network problems or tomcat (%s) is down %d",
               jk_dump_hinfo(&ae->worker->worker_inet_addr, buf), rc);
        JK_TRACE_EXIT(l);
        return JK_FALSE;
    }
    ae->endpoint.rd += rc;

    if (ae->proto == AJP13_PROTO) {
        if (JK_IS_DEBUG_LEVEL(l))
            jk_dump_buff(l, JK_LOG_DEBUG, "received from ajp13", msg);
    }
    else if (ae->proto == AJP14_PROTO) {
        if (JK_IS_DEBUG_LEVEL(l))
            jk_dump_buff(l, JK_LOG_DEBUG, "received from ajp14", msg);
    }
    JK_TRACE_EXIT(l);
    return JK_TRUE;
}

/*
 * Read all the data from the socket.
 *
 * Socket API doesn't guaranty that all the data will be kept in a
 * single read, so we must loop until all awaited data is received
 */

static int ajp_read_fully_from_server(jk_ws_service_t *s, jk_logger_t *l,
                                      unsigned char *buf, unsigned int len)
{
    unsigned int rdlen = 0;
    unsigned int padded_len = len;

    JK_TRACE_ENTER(l);
    if (s->is_chunked && s->no_more_chunks) {
        JK_TRACE_EXIT(l);
        return 0;
    }
    if (s->is_chunked) {
        /* Corner case: buf must be large enough to hold next
         * chunk size (if we're on or near a chunk border).
         * Pad the length to a reasonable value, otherwise the
         * read fails and the remaining chunks are tossed.
         */
        padded_len = (len < CHUNK_BUFFER_PAD) ? len : len - CHUNK_BUFFER_PAD;
    }

    while (rdlen < padded_len) {
        unsigned int this_time = 0;
        if (!s->read(s, buf + rdlen, len - rdlen, &this_time)) {
            /* Remote Client read failed. */
            JK_TRACE_EXIT(l);
            return JK_CLIENT_ERROR;
        }

        if (0 == this_time) {
            if (s->is_chunked) {
                s->no_more_chunks = 1;  /* read no more */
            }
            break;
        }
        rdlen += this_time;
    }

    return (int)rdlen;
}


/*
 * Read data from AJP13/AJP14 protocol
 * Returns -1 on error, else number of bytes read
 */

static int ajp_read_into_msg_buff(ajp_endpoint_t * ae,
                                  jk_ws_service_t *r,
                                  jk_msg_buf_t *msg, int len, jk_logger_t *l)
{
    unsigned char *read_buf = msg->buf;

    JK_TRACE_ENTER(l);
    jk_b_reset(msg);

    read_buf += AJP_HEADER_LEN; /* leave some space for the buffer headers */
    read_buf += AJP_HEADER_SZ_LEN;      /* leave some space for the read length */

    /* Pick the max size since we don't know the content_length */
    if (r->is_chunked && len == 0) {
        len = AJP13_MAX_SEND_BODY_SZ;
    }

    if ((len = ajp_read_fully_from_server(r, l, read_buf, len)) < 0) {
        jk_log(l, JK_LOG_INFO,
               "Receiving data from client failed. "
               "Connection aborted or network problems");
        JK_TRACE_EXIT(l);
        return JK_CLIENT_ERROR;
    }

    if (!r->is_chunked) {
        ae->left_bytes_to_send -= len;
    }

    if (len > 0) {
        /* Recipient recognizes empty packet as end of stream, not
           an empty body packet */
        if (0 != jk_b_append_int(msg, (unsigned short)len)) {
            jk_log(l, JK_LOG_INFO,
                   "Failed appending message length");
            JK_TRACE_EXIT(l);
            return JK_CLIENT_ERROR;
        }
    }

    msg->len += len;

    JK_TRACE_EXIT(l);
    return len;
}


/*
 * send request to Tomcat via Ajp13
 * - first try to find reuseable socket
 * - if no one available, try to connect
 * - send request, but send must be see as asynchronous,
 *   since send() call will return noerror about 95% of time
 *   Hopefully we'll get more information on next read.
 *
 * nb: reqmsg is the original request msg buffer
 *     repmsg is the reply msg buffer which could be scratched
 */
static int ajp_send_request(jk_endpoint_t *e,
                            jk_ws_service_t *s,
                            jk_logger_t *l,
                            ajp_endpoint_t * ae, ajp_operation_t * op)
{
    int err = 0;
    int postlen;

    JK_TRACE_ENTER(l);
    /* Up to now, we can recover */
    op->recoverable = JK_TRUE;

    /*
     * First try to reuse open connections...
     */
    while ((ae->sd > 0)) {
        int rc = 0;
        err = 0;
        if (ae->worker->socket_timeout) {
            if (!jk_is_socket_connected(ae->sd)) {
                jk_log(l, JK_LOG_INFO,
                       "Socket %d is not connected any more (errno=%d)",
                       ae->sd, errno);
                jk_close_socket(ae->sd);
                ae->sd = -1;
                err++;
            }
        }
        if (ae->worker->prepost_timeout != 0 && !err) {
            /* handle cping/cpong if prepost_timeout is set
             * If the socket is disconnected no need to handle
             * the cping/cpong
             */
            if (ajp_handle_cping_cpong(ae, ae->worker->prepost_timeout, l) ==
                JK_FALSE)
                err++;
        }

        /* If we got an error or can't send data, then try to get a pooled
         * connection and try again.  If we are succesful, break out of this
         * loop. */
        if (err ||
            ((rc = ajp_connection_tcp_send_message(ae, op->request, l)) != JK_TRUE)) {
            if (rc != JK_FATAL_ERROR) {
                jk_log(l, JK_LOG_INFO,
                       "Error sending request. Will try another pooled connection");
                ajp_next_connection(ae, l);
            }
            else {
                op->recoverable = JK_FALSE;
                jk_log(l, JK_LOG_INFO,
                       "Error sending request. Unrecoverable operation");
                JK_TRACE_EXIT(l);
                return JK_FALSE;
            }
        }
        else
            break;
    }

    /*
     * If we failed to reuse a connection, try to reconnect.
     */
    if (ae->sd < 0) {

        if (err) {
            /* XXX: If err is set, the tomcat is either dead or disconnected */
            jk_log(l, JK_LOG_INFO,
                   "All endpoints are disconnected or dead");
            JK_TRACE_EXIT(l);
            return JK_FALSE;
        }
        /* no need to handle cping/cpong here since it should be at connection time */

        if (ajp_connect_to_endpoint(ae, l) == JK_TRUE) {
            /*
             * After we are connected, each error that we are going to
             * have is probably unrecoverable
             */
            if (ajp_connection_tcp_send_message(ae, op->request, l) != JK_TRUE) {
                jk_log(l, JK_LOG_INFO,
                       "Error sending request on a fresh connection");
                JK_TRACE_EXIT(l);
                return JK_FALSE;
            }
        }
        else {
            /* Close the socket if unable to connect */