diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 123f11d5..14286a21 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -209,7 +209,7 @@ template class WFNetworkTask : public CommRequest { public: - /* start(), dismiss() for client task only. */ + /* start(), dismiss() for client tasks only. */ void start() { assert(!series_of(this)); @@ -222,13 +222,18 @@ public: delete this; } - /* noreply() for server task only. */ + /* noreply(), push() for server tasks only. */ void noreply() { if (this->state == WFT_STATE_TOREPLY) this->state = WFT_STATE_NOREPLY; } + int push(const void *buf, size_t size) + { + return this->scheduler->push(buf, size, this); + } + public: REQ *get_req() { return &this->req; } RESP *get_resp() { return &this->resp; } diff --git a/src/kernel/CommScheduler.h b/src/kernel/CommScheduler.h index a4951a13..5a85ea9a 100644 --- a/src/kernel/CommScheduler.h +++ b/src/kernel/CommScheduler.h @@ -144,6 +144,11 @@ public: return this->comm.reply(session); } + int push(const void *buf, size_t size, CommSession *session) + { + return this->comm.push(buf, size, session); + } + int bind(CommService *service) { return this->comm.bind(service); diff --git a/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index a83b220e..99e5d042 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -157,7 +157,6 @@ void CommTarget::deinit() int CommMessageIn::feedback(const void *buf, size_t size) { struct CommConnEntry *entry = this->entry; - int error; int ret; if (!entry->ssl) @@ -169,9 +168,9 @@ int CommMessageIn::feedback(const void *buf, size_t size) ret = SSL_write(entry->ssl, buf, size); if (ret <= 0) { - error = SSL_get_error(entry->ssl, ret); - if (error != SSL_ERROR_SYSCALL) - errno = -error; + ret = SSL_get_error(entry->ssl, ret); + if (ret != SSL_ERROR_SYSCALL) + errno = -ret; ret = -1; } @@ -1659,6 +1658,49 @@ int Communicator::reply(CommSession *session) return 0; } +int Communicator::push(const void *buf, size_t size, CommSession *session) +{ + CommTarget *target = session->target; + struct CommConnEntry *entry; + int ret; + + if (session->passive != 1) + { + errno = session->passive ? ENOENT : EINVAL; + return -1; + } + + pthread_mutex_lock(&target->mutex); + if (!list_empty(&target->idle_list)) + { + entry = list_entry(target->idle_list.next, struct CommConnEntry, list); + if (!entry->ssl) + ret = write(entry->sockfd, buf, size); + else if (size == 0) + ret = 0; + else + { + ret = SSL_write(entry->ssl, buf, size); + if (ret <= 0) + { + ret = SSL_get_error(entry->ssl, ret); + if (ret != SSL_ERROR_SYSCALL) + errno = -ret; + + ret = -1; + } + } + } + else + { + errno = ENOENT; + ret = -1; + } + + pthread_mutex_unlock(&target->mutex); + return ret; +} + int Communicator::sleep(SleepSession *session) { struct timespec value; diff --git a/src/kernel/Communicator.h b/src/kernel/Communicator.h index 6f101a32..3b6e09b7 100644 --- a/src/kernel/Communicator.h +++ b/src/kernel/Communicator.h @@ -275,6 +275,8 @@ public: int request(CommSession *session, CommTarget *target); int reply(CommSession *session); + int push(const void *buf, size_t size, CommSession *session); + int bind(CommService *service); void unbind(CommService *service);