update code

src/kernel/Communicator.cc

@@ -314,11 +314,6 @@ private:
 	friend class Communicator;
 };
 
-CommConnection *CommSession::get_connection() const
-{
-	return this->entry ? entry->conn : NULL;
-}
-
 CommSession::~CommSession()
 {
 	struct CommConnEntry *entry;
@@ -363,6 +358,18 @@ CommMessageOut *CommChannel::message_out()
 	return ∅
 }
 
+CommMessageOut *CommSessionOut::message_out()
+{
+	errno = EPERM;
+	return NULL;
+}
+
+CommMessageIn *CommSessionOut::message_in()
+{
+	errno = EPERM;
+	return NULL;
+}
+
 inline int Communicator::first_timeout(CommSession *session)
 {
 	int timeout = session->target->response_timeout;
@@ -1225,7 +1232,7 @@ int Communicator::create_service_session(struct CommConnEntry *entry)
 		session->passive = 1;
 		entry->session = session;
 		session->target = target;
-		session->entry = entry;
+		session->conn = entry->conn;
 		session->seq = entry->seq++;
 		session->out = NULL;
 		session->in = NULL;
@@ -1359,7 +1366,7 @@ void Communicator::callback(struct poller_result *res, void *context)
 			}
 
 			free(entry);
-			entry = session->entry;
+			entry = ((CommSessionOut *)session)->entry;
 			session->handle(state, res->error);
 			session = entry->session;
 		}
@@ -1622,7 +1629,7 @@ int Communicator::request_idle_conn(CommSession *session, CommTarget *target)
 	if (entry)
 	{
 		entry->session = session;
-		session->entry = entry;
+		session->conn = entry->conn;
 		session->seq = entry->seq++;
 		session->out = session->message_out();
 		if (session->out)
@@ -1665,7 +1672,7 @@ int Communicator::request(CommSession *session, CommTarget *target)
 		entry = this->launch_conn(session, target);
 		if (entry)
 		{
-			session->entry = entry;
+			session->conn = entry->conn;
 			session->seq = entry->seq++;
 			data.operation = PD_OP_CONNECT;
 			data.fd = entry->sockfd;
@@ -1677,7 +1684,7 @@ int Communicator::request(CommSession *session, CommTarget *target)
 			this->release_conn(entry);
 		}
 
-		session->entry = NULL;
+		session->conn = NULL;
 		session->seq = 0;
 		return -1;
 	}
@@ -1838,15 +1845,7 @@ int Communicator::establish(CommChannel *channel, CommTarget *target)
 	return -1;
 }
 
-void Communicator::prep_send(CommSession *session, CommChannel *channel)
-{
-	session->target = channel->target;
-	session->entry = channel->entry;
-	session->in = NULL;
-	session->seq = 0;
-}
-
-int Communicator::send(CommMessageOut *msg, CommSession *session,
+int Communicator::send(CommMessageOut *msg, CommSessionOut *session,
 					   CommChannel *channel)
 {
 	struct CommConnEntry *entry = channel->entry;
@@ -1854,6 +1853,11 @@ int Communicator::send(CommMessageOut *msg, CommSession *session,
 	struct iovec *end;
 	int cnt;
 
+	session->target = channel->target;
+	session->entry = channel->entry;
+	session->in = NULL;
+	session->seq = 0;
+
 	session->out = msg;
 	cnt = msg->encode(vectors, ENCODE_IOV_MAX);
 	if ((unsigned int)cnt > ENCODE_IOV_MAX)
@@ -1867,8 +1871,10 @@ int Communicator::send(CommMessageOut *msg, CommSession *session,
 	if (!entry->ssl || cnt == 0)
 	{
 		cnt = __send_vectors(vectors, cnt, entry);
-		if (cnt <= 0)
-			return cnt;
+		if (cnt == 0)
+			return 1;
+		else if (cnt < 0)
+			return -1;
 	}
 
 	entry = (struct CommConnEntry *)malloc(sizeof (struct CommConnEntry));
@@ -1884,7 +1890,7 @@ int Communicator::send(CommMessageOut *msg, CommSession *session,
 			if (this->send_message_async(end - cnt, cnt, entry) >= 0)
 			{
 				__sync_add_and_fetch(&channel->entry->ref, 1);
-				return 1;
+				return 0;
 			}
 
 			close(entry->sockfd);

src/kernel/Communicator.h

@@ -134,14 +134,14 @@ private:
 
 protected:
 	CommTarget *get_target() const { return this->target; }
+	CommConnection *get_connection() const { return this->conn; }
 	CommMessageOut *get_message_out() const { return this->out; }
 	CommMessageIn *get_message_in() const { return this->in; }
 	long long get_seq() const { return this->seq; }
-	CommConnection *get_connection() const;
 
 private:
 	CommTarget *target;
-	struct CommConnEntry *entry;
+	CommConnection *conn;
 	CommMessageOut *out;
 	CommMessageIn *in;
 	long long seq;
@@ -243,8 +243,22 @@ private:
 
 private:
 	virtual CommMessageOut *message_out(); /* final */
-	virtual CommMessageOut *get_message_out() { return NULL; } /* deleted */
+	CommMessageOut *get_message_out() { return NULL; } /* deleted */
 
+private:
+	struct CommConnEntry *entry;
+	friend class Communicator;
+};
+
+class CommSessionOut : public CommSession
+{
+private:
+	virtual CommMessageOut *message_out(); /* final */
+	virtual CommMessageIn *message_in(); /* final */
+	CommMessageIn *get_message_in() { return NULL; } /* deleted */
+
+private:
+	struct CommConnEntry *entry;
 	friend class Communicator;
 };
 
@@ -282,8 +296,8 @@ public:
 	void unbind(CommService *service);
 
 	int establish(CommChannel *channel, CommTarget *target);
-	void prep_send(CommSession *session, CommChannel *channel);
-	int send(CommMessageOut *msg, CommSession *session, CommChannel *channel);
+	int send(CommMessageOut *msg, CommSessionOut *session,
+			 CommChannel *channel);
 	void shutdown(CommChannel *channel);
 
 	int sleep(SleepSession *session);

tutorial/CMakeLists.txt

@@ -23,6 +23,7 @@ else ()
 endif ()
 
 set(TUTORIAL_LIST
+	test_channel
 	tutorial-01-wget
 	tutorial-02-redis_cli
 	tutorial-03-wget_to_redis