rowproc-demos.diff

diff --git a/src/test/examples/Makefile b/src/test/examples/Makefile
index bbc6ee1..e0d6c41 100644
--- a/src/test/examples/Makefile
+++ b/src/test/examples/Makefile
@@ -14,9 +14,11 @@ override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS)
 override LDLIBS := $(libpq_pgport) $(LDLIBS)
 
 
-PROGS = testlibpq testlibpq2 testlibpq3 testlibpq4 testlo
+PROGS = testlibpq testlibpq2 testlibpq3 testlibpq4 testlo \
+	rowproc-sync rowproc-async getrow-sync getrow-async
 
 all: $(PROGS)
 
 clean:
-	rm -f $(PROGS)
+	rm -f $(PROGS) *.o
+
diff --git a/src/test/examples/getrow-async.c b/src/test/examples/getrow-async.c
new file mode 100644
index 0000000..d74f77a
--- /dev/null
+++ b/src/test/examples/getrow-async.c
@@ -0,0 +1,196 @@
+/*
+ * PQgetRow async demo.
+ *
+ * usage: getrow-async [connstr [query]]
+ */
+
+#include <sys/select.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <libpq-fe.h>
+
+struct Context {
+	PGconn *db;
+	int count;
+};
+
+static void die(PGconn *db, const char *msg)
+{
+	fprintf(stderr, "%s: %s", msg, PQerrorMessage(db));
+	exit(1);
+}
+
+/* wait for event on socket */
+static void db_wait(PGconn *db, int for_write)
+{
+	int fd = PQsocket(db);
+	fd_set fds;
+	int res;
+
+retry:
+	FD_ZERO(&fds);
+	FD_SET(fd, &fds);
+	if (for_write)
+		res = select(fd+1, NULL, &fds, NULL, NULL);
+	else
+		res = select(fd+1, &fds, NULL, NULL, NULL);
+
+	if (res == 0)
+		goto retry;
+	if (res < 0 && errno == EINTR)
+		goto retry;
+	if (res < 0)
+	{
+		fprintf(stderr, "select() failed: %s", strerror(errno));
+		exit(1);
+	}
+}
+
+static void proc_row(struct Context *ctx, PGresult *res)
+{
+	const char *val = PQgetvalue(res, 0, 0);
+	ctx->count++;
+	if (0)
+	printf("column#0: %s\n", val ? val : "NULL");
+}
+
+static void proc_result(struct Context *ctx, PGresult *r)
+{
+	ExecStatusType s;
+
+	s = PQresultStatus(r);
+	if (s == PGRES_TUPLES_OK)
+		printf("query successful, got %d rows\n", ctx->count);
+	else
+		printf("%s: %s\n", PQresStatus(s), PQerrorMessage(ctx->db));
+	PQclear(r);
+}
+
+/*
+ * Handle socket read event
+ *
+ * Returns:
+ * -1 - error
+ *  0 - need to read more data
+ *  1 - all done
+ */
+
+static int socket_read_cb(struct Context *ctx)
+{
+	PGresult *r;
+
+	/* read incoming data */
+	if (!PQconsumeInput(ctx->db))
+		return -1;
+
+	/*
+	 * One query may result in several PGresults,
+	 * first loop is over all PGresults.
+	 */
+	while (1) {
+		/*
+		 * Process all rows already in buffer.
+		 */
+		while (1) {
+			r = PQgetRow(ctx->db);
+			if (!r)
+				break;
+
+			proc_row(ctx, r);
+
+			PQclear(r);
+		}
+
+		/* regular async logic follows */
+
+		/* Need more data from network */
+		if (PQisBusy(ctx->db))
+			return 0;
+
+		/* we have complete PGresult ready */
+		r = PQgetResult(ctx->db);
+		if (r == NULL) {
+			/* all results have arrived */
+			return 1;
+		} else {
+			/* process final resultset status */
+			proc_result(ctx, r);
+		}
+	}
+}
+
+static void exec_query(struct Context *ctx, const char *q)
+{
+	int res;
+	int waitWrite;
+	PGconn *db = ctx->db;
+
+	ctx->count = 0;
+
+	/* launch query */
+	if (!PQsendQuery(ctx->db, q))
+		die(ctx->db, "PQsendQuery");
+
+	/* flush query */
+	res = PQflush(db);
+	if (res < 0)
+		die(db, "flush 1");
+	waitWrite = res > 0;
+
+	/* read data */
+	while (1) {
+		/* sleep until event */
+		db_wait(ctx->db, waitWrite);
+
+		/* got event, process it */
+		if (waitWrite) {
+			/* still more to flush? */
+			res = PQflush(db);
+			if (res < 0)
+				die(db, "flush 2");
+			waitWrite = res > 0;
+		} else {
+			/* read result */
+			res = socket_read_cb(ctx);
+			if (res < 0)
+				die(db, "socket_read_cb");
+			if (res > 0)
+				return;
+			waitWrite = 0;
+		}
+	}
+}
+
+int main(int argc, char *argv[])
+{
+	const char *connstr;
+	const char *q;
+	PGconn *db;
+	struct Context ctx;
+
+	connstr = "dbname=postgres";
+	if (argc > 1)
+		connstr = argv[1];
+
+	q = "show all";
+	if (argc > 2)
+		q = argv[2];
+
+	db = PQconnectdb(connstr);
+	if (!db || PQstatus(db) == CONNECTION_BAD)
+		die(db, "connect");
+
+	/* set up socket */
+	PQsetnonblocking(db, 1);
+
+	ctx.db = db;
+	exec_query(&ctx, q);
+
+	PQfinish(db);
+
+	return 0;
+}
+
diff --git a/src/test/examples/getrow-sync.c b/src/test/examples/getrow-sync.c
new file mode 100644
index 0000000..a7ed4f6
--- /dev/null
+++ b/src/test/examples/getrow-sync.c
@@ -0,0 +1,83 @@
+/*
+ * PQgetRow sync demo.
+ *
+ * usage: getrow-sync [connstr [query]]
+ */
+
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <libpq-fe.h>
+
+struct Context {
+	PGconn *db;
+	int count;
+};
+
+static void die(PGconn *db, const char *msg)
+{
+	fprintf(stderr, "%s: %s\n", msg, PQerrorMessage(db));
+	exit(1);
+}
+
+static void exec_query(struct Context *ctx, const char *q)
+{
+	PGconn *db = ctx->db;
+	PGresult *r;
+	ExecStatusType s;
+
+	ctx->count = 0;
+
+	if (!PQsendQuery(db, q))
+		die(db, "PQsendQuery");
+
+	/* loop with PQgetRow until final PGresult is available */
+	while (1) {
+		r = PQgetRow(db);
+		if (!r)
+			break;
+		ctx->count++;
+		PQclear(r);
+	}
+
+	/* final PGresult, either PGRES_TUPLES_OK or error */
+	r = PQgetResult(db);
+	s = PQresultStatus(r);
+	if (s == PGRES_TUPLES_OK)
+		printf("query successful, got %d rows\n", ctx->count);
+	else
+		printf("%s: %s\n", PQresStatus(s), PQerrorMessage(db));
+
+	PQclear(r);
+}
+
+
+int main(int argc, char *argv[])
+{
+	const char *connstr;
+	const char *q = "show all";
+	PGconn *db;
+	struct Context ctx;
+
+	connstr = "dbname=postgres";
+	if (argc > 1)
+		connstr = argv[1];
+
+	q = "show all";
+	if (argc > 2)
+		q = argv[2];
+
+	db = PQconnectdb(connstr);
+	if (!db || PQstatus(db) == CONNECTION_BAD)
+		die(db, "connect");
+
+	ctx.db = db;
+	exec_query(&ctx, q);
+
+	PQfinish(db);
+
+	return 0;
+}
+
diff --git a/src/test/examples/rowproc-async.c b/src/test/examples/rowproc-async.c
new file mode 100644
index 0000000..88b1672
--- /dev/null
+++ b/src/test/examples/rowproc-async.c
@@ -0,0 +1,189 @@
+/*
+ * Row processor async demo.
+ *
+ * usage: rowproc-async [connstr [query]]
+ */
+
+
+#include <sys/select.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <libpq-fe.h>
+
+struct Context {
+	PGconn *db;
+	int count;
+};
+
+/* print db error message and die */
+static void die(PGconn *db, const char *msg)
+{
+	fprintf(stderr, "%s: %s", msg, PQerrorMessage(db));
+	exit(1);
+}
+
+/* wait for event on socket */
+static void db_wait(PGconn *db, int for_write)
+{
+	int fd = PQsocket(db);
+	fd_set fds;
+	int res;
+
+retry:
+	FD_ZERO(&fds);
+	FD_SET(fd, &fds);
+	if (for_write)
+		res = select(fd+1, NULL, &fds, NULL, NULL);
+	else
+		res = select(fd+1, &fds, NULL, NULL, NULL);
+
+	if (res == 0)
+		goto retry;
+	if (res < 0 && errno == EINTR)
+		goto retry;
+	if (res < 0)
+	{
+		fprintf(stderr, "select() failed: %s", strerror(errno));
+		exit(1);
+	}
+}
+
+/* do something with one row */
+static void proc_row(struct Context *ctx, PGresult *res, PGrowValue *columns)
+{
+	ctx->count++;
+
+	if (0)
+	printf("column: %.*s\n",
+		   columns[0].len,
+		   columns[0].value);
+}
+
+/* do something with resultset final status */
+static void proc_result(struct Context *ctx, PGresult *r)
+{
+	ExecStatusType s;
+
+	s = PQresultStatus(r);
+	if (s == PGRES_TUPLES_OK)
+		printf("query successful, got %d rows\n", ctx->count);
+	else
+		printf("%s: %s\n", PQresStatus(s), PQerrorMessage(ctx->db));
+	PQclear(r);
+}
+
+/* custom callback */
+static int my_handler(PGresult *res, PGrowValue *columns, void *arg)
+{
+	struct Context *ctx = arg;
+
+	proc_row(ctx, res, columns);
+
+	return 1;
+}
+
+/* this handles socket read event */
+static int socket_read_cb(struct Context *ctx, PGconn *db)
+{
+	PGresult *r;
+
+	/* read incoming data */
+	if (!PQconsumeInput(db))
+		return -1;
+
+	/*
+	 * one query may result in several PGresult's,
+	 * wrap everything in one big loop.
+	 */
+	while (1) {
+		/* need to wait for more data from network */
+		if (PQisBusy(db))
+			return 0;
+
+		/* we have complete PGresult ready */
+		r = PQgetResult(db);
+		if (r == NULL) {
+			/* all results have arrived */
+			return 1;
+		} else {
+			proc_result(ctx, r);
+		}
+	}
+}
+
+/* run query with custom callback */
+static void exec_query(struct Context *ctx, PGconn *db, const char *q)
+{
+	int res;
+	int waitWrite;
+
+	ctx->count = 0;
+
+	/* set up socket */
+	PQsetnonblocking(db, 1);
+
+	PQsetRowProcessor(db, my_handler, ctx);
+
+	/* launch query */
+	if (!PQsendQuery(db, q))
+		die(db, "PQsendQuery");
+
+	/* see if it is sent */
+	res = PQflush(db); // -1:err, 0:ok, 1:more
+	if (res < 0)
+		die(db, "flush 1");
+	waitWrite = res > 0;
+
+	/* read data */
+	while (1) {
+		db_wait(db, waitWrite);
+
+		/* got event, process it */
+		if (waitWrite) {
+			res = PQflush(db); // -1:err, 0:ok, 1:more
+			if (res < 0)
+				die(db, "flush 2");
+			waitWrite = res > 0;
+		} else {
+			res = socket_read_cb(ctx, db);
+			if (res < 0)
+				die(db, "socket_read_cb");
+			if (res > 0)
+				return;
+			waitWrite = 0;
+		}
+	}
+
+	PQsetRowProcessor(ctx->db, NULL, NULL);
+}
+
+int main(int argc, char *argv[])
+{
+	const char *connstr;
+	const char *q;
+	PGconn *db;
+	struct Context ctx;
+
+	connstr = "dbname=postgres";
+	if (argc > 1)
+		connstr = argv[1];
+
+	q = "show all";
+	if (argc > 2)
+		q = argv[2];
+
+	db = PQconnectdb(connstr);
+	if (!db || PQstatus(db) == CONNECTION_BAD)
+		die(db, "connect");
+	ctx.db = db;
+
+	exec_query(&ctx, db, q);
+
+	PQfinish(db);
+
+	return 0;
+}
+
diff --git a/src/test/examples/rowproc-sync.c b/src/test/examples/rowproc-sync.c
new file mode 100644
index 0000000..b29db48
--- /dev/null
+++ b/src/test/examples/rowproc-sync.c
@@ -0,0 +1,80 @@
+/*
+ * Row processor sync demo.
+ *
+ * usage: rowproc-sync [connstr [query]]
+ */
+
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdarg.h>
+#include <string.h>
+#include <setjmp.h>
+
+#include <libpq-fe.h>
+
+struct Context {
+	PGconn *db;
+	int count;
+};
+
+static void die(PGconn *db, const char *msg)
+{
+	fprintf(stderr, "%s: %s", msg, PQerrorMessage(db));
+	exit(1);
+}
+
+static int my_handler(PGresult *res, PGrowValue *columns, void *arg)
+{
+	struct Context *ctx = arg;
+
+	ctx->count++;
+
+	return 1;
+}
+
+static void exec_query(struct Context *ctx, const char *q)
+{
+	PGresult *r;
+
+	ctx->count = 0;
+	PQsetRowProcessor(ctx->db, my_handler, ctx);
+
+	r = PQexec(ctx->db, q);
+
+	/* check final result */
+	if (!r || PQresultStatus(r) != PGRES_TUPLES_OK)
+		die(ctx->db, "select");
+	else
+		printf("query successful, got %d rows\n", ctx->count);
+	PQclear(r);
+
+	PQsetRowProcessor(ctx->db, NULL, NULL);
+}
+
+
+int main(int argc, char *argv[])
+{
+	const char *connstr;
+	const char *q;
+	struct Context ctx;
+
+	connstr = "dbname=postgres";
+	if (argc > 1)
+		connstr = argv[1];
+
+	q = "show all";
+	if (argc > 2)
+		q = argv[2];
+
+	ctx.db = PQconnectdb(connstr);
+	if (!ctx.db || PQstatus(ctx.db) == CONNECTION_BAD)
+		die(ctx.db, "connect");
+
+	exec_query(&ctx, q);
+
+	PQfinish(ctx.db);
+
+	return 0;
+}
+