add server task's sync push

This commit is contained in:
XieHan
2021-08-29 02:37:59 +08:00
parent 688c7dd85d
commit 25a1b81eca
4 changed files with 60 additions and 6 deletions
+7 -2
View File
@@ -209,7 +209,7 @@ template<class REQ, class RESP>
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; }
+5
View File
@@ -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);
+46 -4
View File
@@ -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;
+2
View File
@@ -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);