client: return requested future with result argument of wait

Currently, the argument returns any decoded future - that is
inconvenient and completely unusable. Let's return only the requested
future, or nothing, if it's not ready.

Along the way, reformat argument list of modified functions to make them
conform clang-format.

Closes #112

Author: drewdzzz

src/Client/Connection.hpp

@@ -230,8 +230,7 @@ class Connection
 
 	template<class B, class N>
 	friend
-	enum DecodeStatus processResponse(Connection<B, N> &conn,
-					  Response<B> *result);
+	enum DecodeStatus processResponse(Connection<B, N> &conn, int req_sync, Response<B> *result);
 
 	template<class B, class N>
 	friend
@@ -530,8 +529,7 @@ inputBufGC(Connection<BUFFER, NetProvider> &conn)
 
 template<class BUFFER, class NetProvider>
 DecodeStatus
-processResponse(Connection<BUFFER, NetProvider> &conn,
-		Response<BUFFER> *result)
+processResponse(Connection<BUFFER, NetProvider> &conn, int req_sync, Response<BUFFER> *result)
 {
 	//Decode response. In case of success - fill in feature map
 	//and adjust end-of-decoded data pointer. Call GC if needed.
@@ -563,7 +561,7 @@ processResponse(Connection<BUFFER, NetProvider> &conn,
 	}
 	LOG_DEBUG("Header: sync=", response.header.sync, ", code=",
 		  response.header.code, ", schema=", response.header.schema_id);
-	if (result != nullptr) {
+	if (result != nullptr && response.header.sync == req_sync) {
 		*result = std::move(response);
 	} else {
 		conn.impl->futures.insert({response.header.sync,

src/Client/Connector.hpp

@@ -94,10 +94,11 @@ class Connector
 	/**
 	 * A helper to decode responses of a connection.
 	 * Can be called when the connection is not ready to decode - it's just no-op.
+	 * If `result` is not `nullptr`, it is used to return response for a request with
+	 * `req_sync` sync. If `result` is `nullptr` - `req_sync` is ignored.
 	 * Returns -1 in the case of any error, 0 on success.
 	 */
-	int connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
-				      Response<BUFFER> *result);
+	int connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn, int req_sync, Response<BUFFER> *result);
 
 private:
 	NetProvider m_NetProvider;
@@ -172,7 +173,7 @@ Connector<BUFFER, NetProvider>::close(ConnectionImpl<BUFFER, NetProvider> &conn)
 
 template<class BUFFER, class NetProvider>
 int
-Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn,
+Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, NetProvider> &conn, int req_sync,
 							  Response<BUFFER> *result)
 {
 	if (!hasDataToDecode(conn))
@@ -183,7 +184,7 @@ Connector<BUFFER, NetProvider>::connectionDecodeResponses(Connection<BUFFER, Net
 
 	int rc = 0;
 	while (hasDataToDecode(conn)) {
-		DecodeStatus status = processResponse(conn,  result);
+		DecodeStatus status = processResponse(conn, req_sync, result);
 		if (status == DECODE_ERR) {
 			rc = -1;
 			break;
@@ -213,9 +214,10 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 	Timer timer{timeout};
 	timer.start();
 	static constexpr int INVALID_SYNC = -1;
+	int req_sync = static_cast<int>(future);
 	if (result != NULL)
 		result->header.sync = INVALID_SYNC;
-	if (connectionDecodeResponses(conn, result) != 0)
+	if (connectionDecodeResponses(conn, req_sync, result) != 0)
 		return -1;
 	if (result != NULL && result->header.sync != INVALID_SYNC) {
 		LOG_DEBUG("Future ", future, " is ready and decoded");
@@ -227,7 +229,7 @@ Connector<BUFFER, NetProvider>::wait(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (connectionDecodeResponses(conn, result) != 0)
+		if (connectionDecodeResponses(conn, req_sync, result) != 0)
 			return -1;
 		if (result != NULL && result->header.sync != INVALID_SYNC) {
 			LOG_DEBUG("Future ", future, " is ready and decoded");
@@ -266,7 +268,7 @@ Connector<BUFFER, NetProvider>::waitAll(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (connectionDecodeResponses(conn, nullptr) != 0)
+		if (connectionDecodeResponses(conn, 0, nullptr) != 0)
 			return -1;
 		bool finish = true;
 		for (size_t i = last_not_ready; i < futures.size(); ++i) {
@@ -306,7 +308,7 @@ Connector<BUFFER, NetProvider>::waitAny(int timeout)
 	}
 	Connection<BUFFER, NetProvider> conn = *m_ReadyToDecode.begin();
 	assert(hasDataToDecode(conn));
-	if (connectionDecodeResponses(conn, nullptr) != 0)
+	if (connectionDecodeResponses(conn, 0, nullptr) != 0)
 		return std::nullopt;
 	return conn;
 }
@@ -325,7 +327,7 @@ Connector<BUFFER, NetProvider>::waitCount(Connection<BUFFER, NetProvider> &conn,
 				      strerror(errno), errno);
 			return -1;
 		}
-		if (connectionDecodeResponses(conn, nullptr) != 0)
+		if (connectionDecodeResponses(conn, 0, nullptr) != 0)
 			return -1;
 		if ((conn.getFutureCount() - ready_futures) >= future_count)
 			return 0;

test/ClientTest.cpp

@@ -1303,6 +1303,27 @@ test_wait(Connector<BUFFER, NetProvider> &client)
 	fail_unless(result.header.sync == static_cast<int>(f));
 	fail_unless(result.header.code == 0);
 
+	TEST_CASE("wait with argument result - several requests");
+	/* Obtain in direct order. */
+	f1 = conn.ping();
+	f2 = conn.ping();
+	fail_unless(client.wait(conn, f1, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f1));
+	fail_unless(result.header.code == 0);
+	fail_unless(client.wait(conn, f2, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f2));
+	fail_unless(result.header.code == 0);
+
+	/* Obtain in reversed order. */
+	f1 = conn.ping();
+	f2 = conn.ping();
+	fail_unless(client.wait(conn, f2, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f2));
+	fail_unless(result.header.code == 0);
+	fail_unless(client.wait(conn, f1, WAIT_TIMEOUT, &result) == 0);
+	fail_unless(result.header.sync == static_cast<int>(f1));
+	fail_unless(result.header.code == 0);
+
 	client.close(conn);
 }