Add length to compression acknolegment message

Author: knizhnik

doc/src/sgml/protocol.sgml

@@ -93,10 +93,10 @@
   </para>
 
   <para>
-    Is is possible to compress protocol data to reduce traffic and speed-up client-server interaction.
-    Compression is especialy useful for importing/exprorting data to/from database using COPY command
-    and for replication (oth physical and logical). Also compression can reduce server response time
-    in case of queries, requestion larger amount of data (for example returning JSON, BLOBs, text,...)
+    It is possible to compress protocol data to reduce traffic and speed-up client-server interaction.
+    Compression is especial useful for importing/exporting data to/from database using COPY command
+    and for replication (both physical and logical). Also compression can reduce server response time
+    in case of queries returning large amount of data (for example returning JSON, BLOBs, text,...)
     Right now compression algorithm is hardcoded: is it is either zlib (default), either zstd (if Postgres was
     configured with --with-zstd option). In both cases streaming mode is used.
   </para>
@@ -3437,7 +3437,16 @@ CompressionOk (B)
   and exchange compressed messages.
 </para>
 </listitem>
-
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Length of message contents in bytes, including self.
+</para>
+</listitem>
 </varlistentry>
 <varlistentry>
 <term>

src/backend/libpq/pqcomm.c

@@ -201,15 +201,14 @@ pq_configure(Port* port)
 {
 	if (port->use_compression)
 	{
-		char compression[2];
+		char compression[6] = {'z',0,0,0,5,0}; /* message length = 5 */
 		int rc;
-		compression[0] = 'z'; /* Request compression message */
-		compression[1] = zpq_algorithm();
+		compression[5] = zpq_algorithm();
 		/* Switch on compression at client side */
 		socket_set_nonblocking(false);
-		while ((rc = secure_write(MyProcPort, &compression, sizeof compression)) < 0
+		while ((rc = secure_write(MyProcPort, compression, sizeof(compression))) < 0
 			   && errno == EINTR);
-		if (rc != 2)
+		if ((size_t)rc != sizeof(compression))
 			return -1;
 
 		/* initialize compression */

src/interfaces/libpq/fe-connect.c

@@ -2880,13 +2880,27 @@ PQconnectPoll(PGconn *conn)
 					if (beresp == 'z') /* Switch on compression */
 					{
 						char algorithm;
+						/* Read message length word */
+						if (pqGetInt(&msgLength, 4, conn))
+						{
+							/* We'll come back when there is more data */
+							return PGRES_POLLING_READING;
+						}
+						if (msgLength != 5)
+						{
+							appendPQExpBuffer(&conn->errorMessage,
+											  libpq_gettext(
+												  "expected compression algorithm specification message length is 5 bytes, but %d is recevied\n"),
+											  msgLength);
+							goto error_return;
+						}
 						pqGetc(&algorithm, conn);
 						if (zpq_algorithm() != algorithm)
 						{
 							appendPQExpBuffer(&conn->errorMessage,
 											  libpq_gettext(
 												  "server and client were configured with different libpq compression algorithms: %c vs. %c\n"),
-											  algorithm, zpq_algorithm());
+										  algorithm, zpq_algorithm());
 							goto error_return;
 						}
 						/* mark byte consumed */

src/interfaces/libpq/fe-misc.c

@@ -61,6 +61,15 @@ static int  pqSocketCheck(PGconn *conn, int forRead, int forWrite,
 						  time_t end_time);
 static int	pqSocketPoll(int sock, int forRead, int forWrite, time_t end_time);
 
+
+#define pq_read_conn(conn,processed)									\
+	(conn->zstream														\
+	 ? zpq_read(conn->zstream, conn->inBuffer + conn->inEnd,			\
+				conn->inBufSize - conn->inEnd, &processed)				\
+	 : pqsecure_read(conn, conn->inBuffer + conn->inEnd,				\
+					 conn->inBufSize - conn->inEnd))
+
+
 /*
  * PQlibVersion: return the libpq version number
  */
@@ -681,11 +690,7 @@ pqReadData(PGconn *conn)
 	/* OK, try to read some data */
 retry3:
 	processed = 0;
-	nread = conn->zstream
-		? zpq_read(conn->zstream, conn->inBuffer + conn->inEnd,
-				   conn->inBufSize - conn->inEnd, &processed)
-		: pqsecure_read(conn, conn->inBuffer + conn->inEnd,
-						conn->inBufSize - conn->inEnd);
+	nread = pq_read_conn(conn,processed);
 	conn->inEnd += processed;
 	if (nread < 0)
 	{
@@ -784,11 +789,7 @@ pqReadData(PGconn *conn)
 	 */
 retry4:
 	processed = 0;
-	nread = conn->zstream
-		? zpq_read(conn->zstream, conn->inBuffer + conn->inEnd,
-				   conn->inBufSize - conn->inEnd, &processed)
-		: pqsecure_read(conn, conn->inBuffer + conn->inEnd,
-						conn->inBufSize - conn->inEnd);
+	nread = pq_read_conn(conn,processed);
 	conn->inEnd += processed;
 
 	if (nread < 0)