Merge pull request #453 from Barenboim/master

make message wrapper compatible with feedback.

src/factory/HttpTaskImpl.cc

@@ -466,7 +466,7 @@ private:
 		SSL *ssl_;
 		SSLHandshaker handshaker_;
 		SSLWrapper wrapper_;
-		SSLConnection(SSL *ssl) : handshaker_(ssl), wrapper_(NULL, ssl)
+		SSLConnection(SSL *ssl) : handshaker_(ssl), wrapper_(&wrapper_, ssl)
 		{
 			ssl_ = ssl;
 		}

src/factory/MySQLTaskImpl.cc

@@ -63,7 +63,7 @@ private:
 	{
 		SSL *ssl_;
 		SSLWrapper wrapper_;
-		SSLConnection(SSL *ssl) : wrapper_(NULL, ssl)
+		SSLConnection(SSL *ssl) : wrapper_(&wrapper_, ssl)
 		{
 			ssl_ = ssl;
 		}

src/kernel/Communicator.h

@@ -110,7 +110,7 @@ private:
 
 protected:
 	/* Send small packet while receiving. Call only in append(). */
-	int feedback(const void *buf, size_t size);
+	virtual int feedback(const void *buf, size_t size);
 
 private:
 	struct CommConnEntry *entry;

src/protocol/ProtocolMessage.h

@@ -33,7 +33,7 @@ namespace protocol
 
 class ProtocolMessage : public CommMessageOut, public CommMessageIn
 {
-public:
+protected:
 	virtual int encode(struct iovec vectors[], int max)
 	{
 		errno = ENOSYS;
@@ -71,17 +71,28 @@ public:
 	void set_attachment(Attachment *att) { this->attachment = att; }
 	Attachment *get_attachment() const { return this->attachment; }
 
+protected:
+	virtual int feedback(const void *buf, size_t size)
+	{
+		if (this->wrapper)
+			return this->wrapper->feedback(buf, size);
+		else
+			return this->CommMessageIn::feedback(buf, size);
+	}
+
 protected:
 	size_t size_limit;
 
 private:
 	Attachment *attachment;
+	ProtocolMessage *wrapper;
 
 public:
 	ProtocolMessage()
 	{
 		this->size_limit = (size_t)-1;
 		this->attachment = NULL;
+		this->wrapper = NULL;
 	}
 
 	virtual ~ProtocolMessage() { delete this->attachment; }
@@ -108,6 +119,32 @@ public:
 
 		return *this;
 	}
+
+	friend class ProtocolWrapper;
+};
+
+class ProtocolWrapper : public ProtocolMessage
+{
+protected:
+	virtual int encode(struct iovec vectors[], int max)
+	{
+		return this->msg->encode(vectors, max);
+	}
+
+	virtual int append(const void *buf, size_t *size)
+	{
+		return this->msg->append(buf, size);
+	}
+
+protected:
+	ProtocolMessage *msg;
+
+public:
+	ProtocolWrapper(ProtocolMessage *msg)
+	{
+		msg->wrapper = this;
+		this->msg = msg;
+	}
 };
 
 }

src/protocol/SSLWrapper.cc

@@ -145,7 +145,7 @@ int SSLWrapper::encode(struct iovec vectors[], int max)
 	if (BIO_reset(wbio) <= 0)
 		return -1;
 
-	ret = this->msg->encode(vectors, max);
+	ret = this->ProtocolWrapper::encode(vectors, max);
 	if ((unsigned int)ret > (unsigned int)max)
 		return ret;
 
@@ -195,7 +195,7 @@ int SSLWrapper::append_message()
 		do
 		{
 			n = nleft;
-			ret = this->msg->append(p, &n);
+			ret = this->ProtocolWrapper::append(p, &n);
 			if (ret == 0)
 			{
 				nleft -= n;
@@ -258,5 +258,34 @@ int ServiceSSLWrapper::append(const void *buf, size_t *size)
 	return -1;
 }
 
+int SSLWrapper::feedback(const void *buf, size_t size)
+{
+	BIO *wbio = SSL_get_wbio(this->ssl);
+	char *ptr;
+	long len;
+	int ret;
+
+	if (size == 0)
+		return 0;
+
+	ret = SSL_write(this->ssl, buf, size);
+	if (ret <= 0)
+	{
+		ret = SSL_get_error(this->ssl, ret);
+		if (ret != SSL_ERROR_SYSCALL)
+			errno = -ret;
+
+		return -1;
+	}
+
+	len = BIO_get_mem_data(wbio, &ptr);
+	if (len > 0)
+		return this->ProtocolWrapper::feedback(ptr, len);
+	else if (len == 0)
+		return 0;
+	else
+		return -1;
+}
+
 }
 

src/protocol/SSLWrapper.h

@@ -42,23 +42,24 @@ public:
 	SSLHandshaker& operator = (SSLHandshaker&& handshaker) = default;
 };
 
-class SSLWrapper : public ProtocolMessage
+class SSLWrapper : public ProtocolWrapper
 {
 protected:
 	virtual int encode(struct iovec vectors[], int max);
 	virtual int append(const void *buf, size_t *size);
 
+protected:
+	virtual int feedback(const void *buf, size_t size);
+
 protected:
 	int append_message();
 
 protected:
-	ProtocolMessage *msg;
 	SSL *ssl;
 
 public:
-	SSLWrapper(ProtocolMessage *msg, SSL *ssl)
+	SSLWrapper(ProtocolMessage *msg, SSL *ssl) : ProtocolWrapper(msg)
 	{
-		this->msg = msg;
 		this->ssl = ssl;
 	}