aboutsummaryrefslogtreecommitdiff
path: root/src/conn/conn.zig
diff options
context:
space:
mode:
authorMartin Ashby <martin@ashbysoft.com>2023-09-27 23:34:46 +0100
committerMartin Ashby <martin@ashbysoft.com>2023-09-27 23:34:46 +0100
commit747c6e55cbe2283fd85ef8cd930e88d2bb0b7db2 (patch)
tree7115e12e19f684640bd2aad4e5d998e13bbb5484 /src/conn/conn.zig
parent08472c27c77d27ea084e3458842540351c5a5c28 (diff)
downloadpgz-747c6e55cbe2283fd85ef8cd930e88d2bb0b7db2.tar.gz
pgz-747c6e55cbe2283fd85ef8cd930e88d2bb0b7db2.tar.bz2
pgz-747c6e55cbe2283fd85ef8cd930e88d2bb0b7db2.tar.xz
pgz-747c6e55cbe2283fd85ef8cd930e88d2bb0b7db2.zip
Add a tagged union for all backend messages.
Move read_message to proto.zig and make it return the tagged union rather than expecting a message type.
Diffstat (limited to 'src/conn/conn.zig')
-rw-r--r--src/conn/conn.zig74
1 files changed, 46 insertions, 28 deletions
diff --git a/src/conn/conn.zig b/src/conn/conn.zig
index 4d62f57..d97eca2 100644
--- a/src/conn/conn.zig
+++ b/src/conn/conn.zig
@@ -2,8 +2,11 @@ const std = @import("std");
const log = std.log.scoped(.pgz);
const SSHashMap = std.StringHashMap([]const u8);
const Config = @import("config.zig");
-const Proto = @import("../proto/proto.zig");
-const read_message = @import("../main.zig").read_message;
+const proto = @import("../proto/proto.zig");
+const StartupMessage = proto.StartupMessage;
+const PasswordMessage = proto.PasswordMessage;
+const BackendMessage = proto.BackendMessage;
+const read_message = proto.read_message;
const ProtocolError = @import("../main.zig").ProtocolError;
const ServerError = @import("../main.zig").ServerError;
const ClientError = @import("../main.zig").ClientError;
@@ -35,63 +38,52 @@ pub fn connect(config: Config) !Conn {
};
errdefer res.deinit();
var writer = stream.writer();
- var dr = diagnosticReader(10000, stream.reader());
+ var dr = diagnosticReader(100, stream.reader());
var reader = dr.reader();
var params = SSHashMap.init(allocator);
try params.put("user", config.user);
if (config.database) |database| try params.put("database", database);
- var sm = Proto.StartupMessage{
+ var sm = StartupMessage{
.parameters = params,
};
defer sm.deinit(allocator);
try sm.write(allocator, writer);
lp: while (true) {
- const response_type = try reader.readByte();
- switch (response_type) {
- Proto.ErrorResponse.Tag => {
- var err = try read_message(Proto.ErrorResponse, allocator, reader);
- defer err.deinit(allocator);
+ var anymsg = try read_message(allocator, reader);
+ defer anymsg.deinit(allocator);
+ switch (anymsg) {
+ .ErrorResponse => |err| {
log.err("Error connecting to server {any}", .{err});
return ServerError.ErrorResponse;
},
- Proto.AuthenticationRequest.Tag => {
- var ar = try read_message(Proto.AuthenticationRequest, allocator, reader);
- defer ar.deinit(allocator);
- // TODO handle the authentication request
+ .AuthenticationRequest => |ar| {
switch (ar.inner_type) {
.AuthRequestTypeOk => {}, // fine do nothing!
.AuthRequestTypeCleartextPassword => {
if (config.password) |password| {
- const pm = Proto.PasswordMessage{ .password = password };
+ const pm = PasswordMessage{ .password = password };
try pm.write(allocator, writer);
} else {
return ClientError.NoPasswordSupplied;
}
},
}
- log.info("authentication request", .{});
},
- Proto.ReadyForQuery.Tag => {
- var rfq = try read_message(Proto.ReadyForQuery, allocator, reader);
- defer rfq.deinit(allocator);
+ .ReadyForQuery => |rfq| {
// TODO do something about transaction state?
res.status = .connStatusIdle;
- log.info("ready for query", .{});
+ log.info("ready for query {any}", .{rfq});
break :lp;
},
- Proto.ParameterStatus.Tag => {
- var ps = try read_message(Proto.ParameterStatus, allocator, reader);
- defer ps.deinit(allocator);
+ .ParameterStatus => |ps| {
// TODO Handle this somehow?
log.info("ParameterStatus: {s}:{s}", .{ ps.name, ps.value });
},
- Proto.BackendKeyData.Tag => {
- var bkd = try read_message(Proto.BackendKeyData, allocator, reader);
- defer bkd.deinit(allocator);
+ .BackendKeyData => |bkd| {
log.info("BackendKeyData process_id {} secret_key {}", .{ bkd.process_id, bkd.secret_key });
},
- else => {
- log.err("unhandled message type [{c}]", .{response_type});
+ else => |response_type| {
+ log.err("unhandled message type [{}]", .{response_type});
const diag = try dr.get(allocator);
defer allocator.free(diag);
log.err("diag [{s}]", .{diag});
@@ -102,10 +94,36 @@ pub fn connect(config: Config) !Conn {
return res;
}
-fn deinit(self: *Conn) void {
+pub fn deinit(self: *Conn) void {
self.stream.close();
}
+// How to handle this ...
+// The Go code relies on polymorphism to generically read any message type.
+// I _could_ have a tagged union type thing
+// pub const ResultIterator = struct {
+// conn: *Conn,
+// command_concluded: bool = false,
+// // NextRow advances the ResultReader to the next row and returns true if a row is available.
+// pub fn next_row(self: *ResultIterator) bool {
+// // TODO implement
+// var reader = self.conn.stream.reader();
+// switch (try reader.readByte()) {
+// case
+// }
+// return false;
+// }
+// };
+
+// pub const MultiResultIterator = struct {
+// conn: *Conn,
+// fn next() ?
+// };
+
+// pub fn exec(self: *Conn) {
+
+// }
+
test "connect unix" {
// must have a local postgres runnning
// TODO maybe use docker to start one?