static apr_status_t pass_data_to_filter(ap_filter_t *f, const char *data,
                                        apr_size_t len, apr_bucket_brigade *bb)
{
    ef_ctx_t *ctx = f->ctx;
    ef_dir_t *dc = ctx->dc;
    apr_status_t rv;
    apr_size_t bytes_written = 0;
    apr_size_t tmplen;

    do {
        tmplen = len - bytes_written;
        rv = apr_file_write(ctx->proc->in,
                       (const char *)data + bytes_written,
                       &tmplen);
        bytes_written += tmplen;
        if (rv != APR_SUCCESS && !APR_STATUS_IS_EAGAIN(rv)) {
            ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, f->r,
                          "apr_file_write(child input), len %" APR_SIZE_T_FMT,
                          tmplen);
            return rv;
        }
        if (APR_STATUS_IS_EAGAIN(rv)) {
            /* XXX handle blocking conditions here...  if we block, we need
             * to read data from the child process and pass it down to the
             * next filter!
             */
            rv = drain_available_output(f, bb);
            if (APR_STATUS_IS_EAGAIN(rv)) {
#if APR_FILES_AS_SOCKETS
                int num_events;
                const apr_pollfd_t *pdesc;

                rv = apr_pollset_poll(ctx->pollset, f->r->server->timeout,
                                      &num_events, &pdesc);
                if (rv || dc->debug >= DBGLVL_GORY) {
                    ap_log_rerror(APLOG_MARK, APLOG_DEBUG,
                                  rv, f->r, "apr_pollset_poll()");
                }
                if (rv != APR_SUCCESS && !APR_STATUS_IS_EINTR(rv)) {
                    /* some error such as APR_TIMEUP */
                    return rv;
                }
#else /* APR_FILES_AS_SOCKETS */
                /* Yuck... I'd really like to wait until I can read
                 * or write, but instead I have to sleep and try again
                 */
                apr_sleep(100000); /* 100 milliseconds */
                if (dc->debug >= DBGLVL_GORY) {
                    ap_log_rerror(APLOG_MARK, APLOG_DEBUG,
                                  0, f->r, "apr_sleep()");
                }
#endif /* APR_FILES_AS_SOCKETS */
            }
            else if (rv != APR_SUCCESS) {
                return rv;
            }
        }
    } while (bytes_written < len);
    return rv;
}

static int proxy_wstunnel_transfer(request_rec *r, conn_rec *c_i, conn_rec *c_o,
                                     apr_bucket_brigade *bb, char *name)
{
    int rv;
#ifdef DEBUGGING
    apr_off_t len;
#endif

    do {
        apr_brigade_cleanup(bb);
        rv = ap_get_brigade(c_i->input_filters, bb, AP_MODE_READBYTES,
                            APR_NONBLOCK_READ, AP_IOBUFSIZE);
        if (rv == APR_SUCCESS) {
            if (c_o->aborted) {
                return APR_EPIPE;
            }
            if (APR_BRIGADE_EMPTY(bb)) {
                break;
            }
#ifdef DEBUGGING
            len = -1;
            apr_brigade_length(bb, 0, &len);
            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, r, APLOGNO(02440)
                          "read %" APR_OFF_T_FMT
                          " bytes from %s", len, name);
#endif
            rv = ap_pass_brigade(c_o->output_filters, bb);
            if (rv == APR_SUCCESS) {
                ap_fflush(c_o->output_filters, bb);
            }
            else {
                ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, APLOGNO(02441)
                              "error on %s - ap_pass_brigade",
                              name);
            }
        } else if (!APR_STATUS_IS_EAGAIN(rv) && !APR_STATUS_IS_EOF(rv)) {
            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, rv, r, APLOGNO(02442)
                          "error on %s - ap_get_brigade",
                          name);
        }
    } while (rv == APR_SUCCESS);

    ap_log_rerror(APLOG_MARK, APLOG_TRACE2, rv, r, "wstunnel_transfer complete");

    if (APR_STATUS_IS_EAGAIN(rv)) {
        rv = APR_SUCCESS;
    }
   
    return rv;
}

static apr_status_t send_brigade_blocking(apr_socket_t *s,
                                          apr_bucket_brigade *bb,
                                          apr_size_t *bytes_written,
                                          conn_rec *c)
{
    apr_status_t rv;

    rv = APR_SUCCESS;
    while (!APR_BRIGADE_EMPTY(bb)) {
        rv = send_brigade_nonblocking(s, bb, bytes_written, c);
        if (rv != APR_SUCCESS) {
            if (APR_STATUS_IS_EAGAIN(rv)) {
                /* Wait until we can send more data */
                apr_int32_t nsds;
                apr_interval_time_t timeout;
                apr_pollfd_t pollset;

                pollset.p = c->pool;
                pollset.desc_type = APR_POLL_SOCKET;
                pollset.reqevents = APR_POLLOUT;
                pollset.desc.s = s;
                apr_socket_timeout_get(s, &timeout);
                rv = apr_poll(&pollset, 1, &nsds, timeout);
                if (rv != APR_SUCCESS) {
                    break;
                }
            }
            else {
                break;
            }
        }
    }
    return rv;
}

apr_status_t h2_mplx_in_read(h2_mplx *m, apr_read_type_e block,
                             int stream_id, apr_bucket_brigade *bb,
                             struct apr_thread_cond_t *iowait)
{
    apr_status_t status; 
    AP_DEBUG_ASSERT(m);
    if (m->aborted) {
        return APR_ECONNABORTED;
    }
    status = apr_thread_mutex_lock(m->lock);
    if (APR_SUCCESS == status) {
        h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
        if (io && !io->orphaned) {
            io->input_arrived = iowait;
            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_pre");
            status = h2_io_in_read(io, bb, -1);
            while (APR_STATUS_IS_EAGAIN(status) 
                   && !is_aborted(m, &status)
                   && block == APR_BLOCK_READ) {
                apr_thread_cond_wait(io->input_arrived, m->lock);
                status = h2_io_in_read(io, bb, -1);
            }
            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_post");
            io->input_arrived = NULL;
        }
        else {
            status = APR_EOF;
        }
        apr_thread_mutex_unlock(m->lock);
    }
    return status;
}

static void do_read(void)
{
    apr_file_t *file;
    apr_status_t status;

    if (apr_file_open(&file, testfile, APR_WRITE,
                 APR_OS_DEFAULT, pool) != APR_SUCCESS)
        errmsg("Could not open test file.\n");
    printf("Test file opened.\n");

    status = apr_file_lock(file, APR_FLOCK_EXCLUSIVE | APR_FLOCK_NONBLOCK);
    if (!APR_STATUS_IS_EAGAIN(status)) {
        char msg[200];
        errmsg(apr_psprintf(pool, "Expected APR_EAGAIN. Got %d: %s.\n",
                            status, apr_strerror(status, msg, sizeof(msg))));
    }
    printf("First attempt: we were properly locked out.\nWaiting for lock...");
    fflush(stdout);

    if (apr_file_lock(file, APR_FLOCK_EXCLUSIVE) != APR_SUCCESS)
        errmsg("Could not establish lock on test file.");
    printf(" got it.\n");

    (void) apr_file_close(file);
    printf("Exiting.\n");
}

/* drain_available_output():
 *
 * if any data is available from the filter, read it and append it
 * to the the bucket brigade
 */
static apr_status_t drain_available_output(ap_filter_t *f,
                                           apr_bucket_brigade *bb)
{
    request_rec *r = f->r;
    conn_rec *c = r->connection;
    ef_ctx_t *ctx = f->ctx;
    ef_dir_t *dc = ctx->dc;
    apr_size_t len;
    char buf[4096];
    apr_status_t rv;
    apr_bucket *b;

    while (1) {
        len = sizeof(buf);
        rv = apr_file_read(ctx->proc->out,
                      buf,
                      &len);
        if ((rv && !APR_STATUS_IS_EAGAIN(rv)) ||
            dc->debug >= DBGLVL_GORY) {
            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, rv, r,
                          "apr_file_read(child output), len %" APR_SIZE_T_FMT,
                          !rv ? len : -1);
        }
        if (rv != APR_SUCCESS) {
            return rv;
        }
        b = apr_bucket_heap_create(buf, len, NULL, c->bucket_alloc);
        APR_BRIGADE_INSERT_TAIL(bb, b);
        return APR_SUCCESS;
    }
    /* we should never get here; if we do, a bogus error message would be
     * the least of our problems
     */
    return APR_ANONYMOUS;
}

static apr_status_t send_out(h2_task *task, apr_bucket_brigade* bb, int block)
{
    apr_off_t written, left;
    apr_status_t status;

    apr_brigade_length(bb, 0, &written);
    H2_TASK_OUT_LOG(APLOG_TRACE2, task, bb, "h2_task send_out");
    h2_beam_log(task->output.beam, task->c, APLOG_TRACE2, "send_out(before)");
    /* engines send unblocking */
    status = h2_beam_send(task->output.beam, bb, 
                          block? APR_BLOCK_READ : APR_NONBLOCK_READ);
    h2_beam_log(task->output.beam, task->c, APLOG_TRACE2, "send_out(after)");
    
    if (APR_STATUS_IS_EAGAIN(status)) {
        apr_brigade_length(bb, 0, &left);
        written -= left;
        status = APR_SUCCESS;
    }
    if (status == APR_SUCCESS) {
        if (h2_task_logio_add_bytes_out) {
            h2_task_logio_add_bytes_out(task->c, written);
        }
        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, task->c, 
                      "h2_task(%s): send_out done", task->id);
    }
    else {
        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, task->c,
                      "h2_task(%s): send_out (%ld bytes)", 
                      task->id, (long)written);
    }
    return status;
}

static apr_status_t serf_response_read(serf_bucket_t *bucket,
                                       apr_size_t requested,
                                       const char **data, apr_size_t *len)
{
    response_context_t *ctx = bucket->data;
    apr_status_t rv;

    rv = wait_for_body(bucket, ctx);
    if (rv) {
        /* It's not possible to have read anything yet! */
        if (APR_STATUS_IS_EOF(rv) || APR_STATUS_IS_EAGAIN(rv)) {
            *len = 0;
        }
        return rv;
    }

    rv = serf_bucket_read(ctx->body, requested, data, len);
    if (APR_STATUS_IS_EOF(rv)) {
        if (ctx->chunked) {
            ctx->state = STATE_TRAILERS;
            /* Mask the result. */
            rv = APR_SUCCESS;
        }
        else {
            ctx->state = STATE_DONE;
        }
    }
    return rv;
}

bool LLPluginProcessParent::accept()
{
    bool result = false;
    apr_status_t status = APR_EGENERAL;

    mSocket = LLSocket::create(status, mListenSocket);

    if(status == APR_SUCCESS)
    {
//		llinfos << "SUCCESS" << llendl;
        // Success.  Create a message pipe on the new socket
        new LLPluginMessagePipe(this, mSocket);

        result = true;
    }
    else
    {
        mSocket.reset();
        // EAGAIN means "No incoming connections". This is not an error.
        if (!APR_STATUS_IS_EAGAIN(status))
        {
            // Some other error.
            ll_apr_warn_status(status);
            errorState();
        }
    }

    return result;
}

static void im_exec_fill_buffer(nx_module_input_t *input)
{
    apr_status_t rv;
    apr_size_t len;

    ASSERT(input != NULL);
    ASSERT(input->buf != NULL);
    ASSERT(input->module != NULL);
    ASSERT(input->desc_type == APR_POLL_FILE);
    ASSERT(input->desc.f != NULL);

    if ( input->bufstart == input->bufsize )
    {
	input->bufstart = 0;
	input->buflen = 0;
    }
    if ( input->buflen == 0 )
    {
	input->bufstart = 0;
    }

    ASSERT(input->bufstart + input->buflen <= input->bufsize);
    len = (apr_size_t) (input->bufsize - (input->buflen + input->bufstart));

    rv = apr_file_read(input->desc.f, input->buf + input->bufstart + input->buflen, &len);

    if ( rv != APR_SUCCESS )
    {
	if ( APR_STATUS_IS_EOF(rv) )
	{
	    throw_msg("Module %s got EOF, process exited? ", input->module->name);
	}
	else if ( APR_STATUS_IS_EAGAIN(rv) )
	{ 
#ifdef WIN32
	    // on windows EAGAIN is normal because we are using NON-BLOCKING reads
	    // so we try again after 500 ms
	    im_exec_add_read_event(input->module, 500);
	    log_debug("got EAGAIN");
#else
	    log_error("got EAGAIN for blocking read in module %s", input->module->name);
#endif
	    ASSERT(len == 0);
	    return;
	}
	else
	{
	    throw(rv, "Module %s couldn't read from pipe", input->module->name);
	}
    }
    else
    {
	log_debug("im_exec read %d bytes", (int) len);
    }
    input->buflen += (int) len;
    ASSERT(input->buflen <= input->bufsize);
}

static int h2_session_status_from_apr_status(apr_status_t rv)
{
    if (rv == APR_SUCCESS) {
        return NGHTTP2_NO_ERROR;
    }
    else if (APR_STATUS_IS_EAGAIN(rv)) {
        return NGHTTP2_ERR_WOULDBLOCK;
    }
    else if (APR_STATUS_IS_EOF(rv)) {
            return NGHTTP2_ERR_EOF;
    }
    return NGHTTP2_ERR_PROTO;
}

bool LLPluginProcessParent::accept()
{
	bool result = false;
	
	apr_status_t status = APR_EGENERAL;
	apr_socket_t *new_socket = NULL;
	
	status = apr_socket_accept(
		&new_socket,
		mListenSocket->getSocket(),
		gAPRPoolp);

	
	if(status == APR_SUCCESS)
	{
//		llinfos << "SUCCESS" << llendl;
		// Success.  Create a message pipe on the new socket

		// we MUST create a new pool for the LLSocket, since it will take ownership of it and delete it in its destructor!
		apr_pool_t* new_pool = NULL;
		status = apr_pool_create(&new_pool, gAPRPoolp);

		mSocket = LLSocket::create(new_socket, new_pool);
		new LLPluginMessagePipe(this, mSocket);

		result = true;
	}
	else if(APR_STATUS_IS_EAGAIN(status))
	{
//		llinfos << "EAGAIN" << llendl;

		// No incoming connections.  This is not an error.
		status = APR_SUCCESS;
	}
	else
	{
//		llinfos << "Error:" << llendl;
		ll_apr_warn_status(status);
		
		// Some other error.
		errorState();
	}
	
	return result;	
}

int jxr_socket_sendfull(jaxer_connection *ac, const unsigned char *b, int len)
{
    jaxer_dir_conf *config = (jaxer_dir_conf *)ap_get_module_config(ac->request->per_dir_config, &jaxer_module);
    int sent = 0;
    apr_socket_t *sock;
	apr_status_t rv;
	apr_size_t send_len;
	int retry = 0;
	apr_pool_t *p = (config) ? config->reqPool : ac->worker->pool;

	sock = ac->sock;

    if (g_jxr_network_trace)
    {
        jxr_trace("SEND", b, len, p);
    }
    while (sent < len)
	{
		send_len = len - sent;
		rv = apr_socket_send(sock, (const char*)(b + sent), &send_len);
		if (send_len == len - sent)
		{
			sent += (int) send_len;
			break;
		}
		if (rv != APR_SUCCESS)
		{
            /*
             * Let's hope this traps EWOULDBLOCK too !
             */
            if (APR_STATUS_IS_EAGAIN(rv) && retry<3)
			{
				retry++;
            }else
			{
				ap_log_perror(APLOG_MARK, APLOG_WARNING, rv, p, "mod_jaxer: send data over socket error: total len=%d sent=%d", len, sent);
				return -1;
			}
        }
		sent += (int) send_len;
	}
	return sent;
}

/*
 * Callback when nghttp2 wants to send bytes back to the client.
 */
static ssize_t send_cb(nghttp2_session *ngh2,
                       const uint8_t *data, size_t length,
                       int flags, void *userp)
{
    h2_session *session = (h2_session *)userp;
    apr_status_t status;
    
    (void)ngh2;
    (void)flags;
    status = h2_conn_io_write(&session->io, (const char *)data, length);
    if (status == APR_SUCCESS) {
        return length;
    }
    if (APR_STATUS_IS_EAGAIN(status)) {
        return NGHTTP2_ERR_WOULDBLOCK;
    }
    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,
                  "h2_session: send error");
    return h2_session_status_from_apr_status(status);
}

/** Receive at max howmany bytes on socket sock and store them in recvbuf.
 *  Parameter howmany allows to receive less than the available space in
 *  recvbuf.
 *  Setting howmany to the same value as recvdbuf->buf_len means receive as
 *  much as you can (offset is taken into consideration).
 */
apr_status_t
cpe_recv(apr_socket_t *sock, cpe_io_buf *recvbuf, apr_size_t *howmany)
{
    apr_status_t  rv;
    apr_size_t    len;

    assert(recvbuf->buf_offset <= recvbuf->buf_capacity);
    if (*howmany <= 0) {
        cpe_log(CPE_DEB, "invalid howmany %d", *howmany);
        return APR_EINVAL;
    }
    len = recvbuf->buf_capacity - recvbuf->buf_offset;
    if (len == 0) {
        cpe_log(CPE_DEB, "no buffer space to receive on buffer %p (socket %p)",
            recvbuf, sock);
        return APR_EGENERAL;
    }
    len = cpe_min(len, *howmany);
    rv = apr_socket_recv(sock, &recvbuf->buf[recvbuf->buf_offset], &len);
    if (rv != APR_SUCCESS) {
        if (APR_STATUS_IS_EAGAIN(rv)) {
            cpe_log(CPE_ERR, "CPE client programming error: socket %p marked "
                "for non-blocking I/O and no data ready to be read.", sock);
        }
        cpe_log(CPE_DEB, "apr_socket_recv: %s (socket %p)", cpe_errmsg(rv),
            sock);
        /* Not clear what to do here; should look at the returned value of
         * len? Should consider APR_EOF ?
         */
    }
    *howmany = len;
    recvbuf->buf_offset += len;
    recvbuf->buf_len = recvbuf->buf_offset;
    recvbuf->total += len;
    cpe_log(CPE_DEB, "received %d bytes (offset %d, socket %p)", len,
        recvbuf->buf_offset, sock);
    return rv;
}

int main(int argc, const char * const *argv)
{
    apr_file_t *file;
    apr_status_t status;
    apr_pool_t *p;

    apr_initialize();
    apr_pool_create(&p, NULL);

    if (apr_file_open(&file, TESTFILE, APR_WRITE, APR_OS_DEFAULT, p) 
        != APR_SUCCESS) {
        
        exit(UNEXPECTED_ERROR);
    }
    status = apr_file_lock(file, APR_FLOCK_EXCLUSIVE | APR_FLOCK_NONBLOCK);
    if (status == APR_SUCCESS) {
        exit(SUCCESSFUL_READ);
    }
    if (APR_STATUS_IS_EAGAIN(status)) {
        exit(FAILED_READ);
    }
    exit(UNEXPECTED_ERROR);
}

int jxr_socket_recvfull(jaxer_connection *ac, unsigned char *b, int len)
{
    jaxer_dir_conf *config = (jaxer_dir_conf *)ap_get_module_config(ac->request->per_dir_config, &jaxer_module);
    int rdlen = 0;
    apr_socket_t *sock;
	apr_status_t rv;
	apr_size_t read_len;
	int retry = 0;
	apr_pool_t *p = (config) ? config->reqPool : ac->worker->pool;

	sock = ac->sock;

    while (rdlen < len)
	{
		read_len = len - rdlen;
		rv = apr_socket_recv(sock, (char *)b + rdlen, &read_len);
		if (rv != APR_SUCCESS)
		{
			if (APR_STATUS_IS_EAGAIN(rv) && retry<3)
			{
				retry++;
            }else
			{
				ap_log_perror(APLOG_MARK, APLOG_WARNING, rv, p, "mod_jaxer: receive data over socket error: total len=%d read=%d", len, rdlen);
				return -1;
			}
		}

        rdlen += (int) read_len;
    }

    if (g_jxr_network_trace)
    {
        jxr_trace("RECV", b, rdlen, p);
    }
    return rdlen;
}

apr_status_t handle_response(serf_request_t *request,
                             serf_bucket_t *response,
                             void *handler_baton,
                             apr_pool_t *pool)
{
    handler_baton_t *ctx = handler_baton;

    if (! response) {
        serf_connection_request_create(ctx->tb->connection,
                                       setup_request,
                                       ctx);
        return APR_SUCCESS;
    }

    while (1) {
        apr_status_t status;
        const char *data;
        apr_size_t len;

        status = serf_bucket_read(response, 2048, &data, &len);
        if (SERF_BUCKET_READ_ERROR(status))
            return status;

        if (APR_STATUS_IS_EOF(status)) {
            APR_ARRAY_PUSH(ctx->handled_requests, int) = ctx->req_id;
            ctx->done = TRUE;
            return APR_EOF;
        }

        if (APR_STATUS_IS_EAGAIN(status)) {
            return status;
        }

    }

    return APR_SUCCESS;
}

apr_status_t h2_mplx_in_read(h2_mplx *m, apr_read_type_e block,
                             int stream_id, apr_bucket_brigade *bb, 
                             apr_table_t *trailers,
                             struct apr_thread_cond_t *iowait)
{
    apr_status_t status; 
    int acquired;
    
    AP_DEBUG_ASSERT(m);
    if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
        h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
        if (io && !io->orphaned) {
            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_pre");
            
            h2_io_signal_init(io, H2_IO_READ, m->stream_timeout, iowait);
            status = h2_io_in_read(io, bb, -1, trailers);
            while (APR_STATUS_IS_EAGAIN(status) 
                   && !is_aborted(m, &status)
                   && block == APR_BLOCK_READ) {
                ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, m->c,
                              "h2_mplx(%ld-%d): wait on in data (BLOCK_READ)", 
                              m->id, stream_id);
                status = h2_io_signal_wait(m, io);
                if (status == APR_SUCCESS) {
                    status = h2_io_in_read(io, bb, -1, trailers);
                }
            }
            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_post");
            h2_io_signal_exit(io);
        }
        else {
            status = APR_EOF;
        }
        leave_mutex(m, acquired);
    }
    return status;
}

static int connection_rx_cb(serv_ctx_t *serv_ctx, apr_pollset_t *pollset, apr_socket_t *sock)
{
    //printf("connection_rx_cb()\n");


    apr_size_t len = BUFSIZE - serv_ctx->down_buf_level;

    apr_status_t rv = apr_socket_recv(sock, (char*)&serv_ctx->down_buf[serv_ctx->down_buf_level], &len); //Daten empfangen
    if (APR_STATUS_IS_EAGAIN(rv))
    {
        /* we have no data to read. we should keep polling the socket */
        return TRUE;
    }
    else if (APR_STATUS_IS_EOF(rv) || len == 0)
    {
        /* we lost TCP session.
         * XXX On Windows, rv would equal to APR_SUCCESS and len==0 in this case. So, we should check @len in addition to APR_EOF check */
        return FALSE;
    }
    else
    {
        /* we got data */
//        printf("rx(%d): ", serv_ctx->channel_number);
//        for(size_t i = 0; i < len ;++i)
//            printf("0x%02x ", (int)serv_ctx->down_buf[serv_ctx->down_buf_level+i]);
//        printf("\n");

        serv_ctx->down_buf_level += len;

        //ipdbgJtagWrite(serv_ctx->chain, (uint8_t*)buf, len, serv_ctx->valid_mask);

    }


    return TRUE;
}