dnrops
/
postgres-wire-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Maintain reference to QueryClientConnection while processing packet

This commit is contained in:
Tommy van der Vorst 2017-11-24 12:23:38 +01:00
parent db9edd5ede
commit 55b8e6e08d
1 changed files with 48 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
Sources/PostgresWireServer/QueryConnection.swift
View File

@ -643,39 +643,44 @@ public final class QueryClientConnection<PreparedStatementType: PreparedStatemen
let queue = DispatchQueue.global(qos: .default) let queue = DispatchQueue.global(qos: .default)
// Create the run loop work item and dispatch to the default priority global queue... // Create the run loop work item and dispatch to the default priority global queue...
queue.async { [unowned self] in queue.async { [weak self] in
var shouldKeepRunning = true if let s = self {
do { var shouldKeepRunning = true
switch self.state { do {
case .new: switch s.state {
if let len = self.readInt32(), let msg = self.readInt32() { case .new:
if len == 8 && msg == 80877103 { if let len = s.readInt32(), let msg = s.readInt32() {
// No SSL, thank you if len == 8 && msg == 80877103 {
try self.socket.write(from: "N") // No SSL, thank you
} try s.socket.write(from: "N")
else if len > UInt32(8) { }
// Read client version number else if len > UInt32(8) {
self.majorVersion = UInt16(msg >> 16) // Read client version number
self.minorVersion = UInt16(msg & 0xFFFF) s.majorVersion = UInt16(msg >> 16)
s.minorVersion = UInt16(msg & 0xFFFF)
// Read parameters // Read parameters
if let p = try self.readParameters(length: len - UInt32(8)) { if let p = try s.readParameters(length: len - UInt32(8)) {
self.username = p["user"] s.username = p["user"]
// Send authentication request // Send authentication request
let buf = Data(bytes: [UInt8(Character("R").codePoint), 0, 0, 0, 8, 0, 0, 0, 3]) let buf = Data(bytes: [UInt8(Character("R").codePoint), 0, 0, 0, 8, 0, 0, 0, 3])
try self.socket.write(from: buf) try s.socket.write(from: buf)
// Read authentication // Read authentication
if let pw = try self.readAuthentication() { if let pw = try s.readAuthentication() {
self.password = pw s.password = pw
// Send authentication success // Send authentication success
let buf = Data(bytes: [UInt8(Character("R").codePoint), 0, 0, 0, 8, 0, 0, 0, 0]) let buf = Data(bytes: [UInt8(Character("R").codePoint), 0, 0, 0, 8, 0, 0, 0, 0])
try self.socket.write(from: buf) try s.socket.write(from: buf)
self.state = .ready s.state = .ready
try self.sendReadyForQuery() try s.sendReadyForQuery()
}
else {
shouldKeepRunning = false
}
} }
else { else {
shouldKeepRunning = false shouldKeepRunning = false
@ -685,30 +690,27 @@ public final class QueryClientConnection<PreparedStatementType: PreparedStatemen
shouldKeepRunning = false shouldKeepRunning = false
} }
} }
else {
case .ready, .querying:
if try !s.readPacket() {
shouldKeepRunning = false shouldKeepRunning = false
} }
}
case .ready, .querying: case .closed:
if try !self.readPacket() { return
shouldKeepRunning = false
} }
case .closed:
return
} }
} catch {
catch { try? s.send(error: error.localizedDescription)
try? self.send(error: error.localizedDescription) shouldKeepRunning = false
shouldKeepRunning = false }
}
if shouldKeepRunning { if shouldKeepRunning {
self.run() s.run()
} }
else { else {
self.close() s.close()
}
} }
} }
} }