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 +#include +#include +#include +#include + +#include + +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 +#include +#include + +#include + +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 +#include +#include +#include +#include + +#include + +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 +#include +#include +#include +#include + +#include + +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; +} +