docker/pgcat
1
0
Fork 0
mirror of https://github.com/postgresml/pgcat.git synced 2026-03-23 01:16:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

COPY OUT support

Browse Source
This commit is contained in:
Lev Kokotov
2022-02-04 08:26:50 -08:00
parent 77507e7343
commit 381f06d46f
2 changed files with 48 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
src/client.rs
View File

@@ -118,14 +118,21 @@ impl Client {
match code { match code {
'Q' => { 'Q' => {
server.send(original).await?; server.send(original).await?;
let response = server.recv().await?;
match write_all_half(&mut self.write, response).await { loop {
Ok(_) => (), let response = server.recv().await?;
Err(err) => { match write_all_half(&mut self.write, response).await {
server.mark_bad(); Ok(_) => (),
return Err(err); Err(err) => {
server.mark_bad();
return Err(err);
}
};
if !server.is_data_available() {
break;
} }
}; }
// Release server // Release server
if !server.in_transaction() { if !server.in_transaction() {
@@ -166,14 +173,20 @@ impl Client {
server.send(self.buffer.clone()).await?; server.send(self.buffer.clone()).await?;
self.buffer.clear(); self.buffer.clear();
let response = server.recv().await?; loop {
match write_all_half(&mut self.write, response).await { let response = server.recv().await?;
Ok(_) => (), match write_all_half(&mut self.write, response).await {
Err(err) => { Ok(_) => (),
server.mark_bad(); Err(err) => {
return Err(err); server.mark_bad();
return Err(err);
}
};
if !server.is_data_available() {
break;
} }
}; }
// Release server // Release server
if !server.in_transaction() { if !server.in_transaction() {

21
src/server.rs
View File

@@ -17,6 +17,7 @@ pub struct Server {
backend_id: i32, backend_id: i32,
secret_key: i32, secret_key: i32,
in_transaction: bool, in_transaction: bool,
data_available: bool,
bad: bool, bad: bool,
} }
@@ -136,6 +137,7 @@ impl Server {
backend_id: backend_id, backend_id: backend_id,
secret_key: secret_key, secret_key: secret_key,
in_transaction: false, in_transaction: false,
data_available: false,
bad: false, bad: false,
}); });
} }
@@ -208,6 +210,21 @@ impl Server {
// CopyInResponse: copy is starting from client to server // CopyInResponse: copy is starting from client to server
'G' => break, 'G' => break,
// CopyOutResponse: copy is starting from the server to the client
'H' => {
self.data_available = true;
break;
}
// CopyData
'd' => break,
// CopyDone
'c' => {
self.data_available = false;
// Buffer until ReadyForQuery shows up
}
_ => { _ => {
// Keep buffering, // Keep buffering,
} }
@@ -224,6 +241,10 @@ impl Server {
self.in_transaction self.in_transaction
} }
pub fn is_data_available(&self) -> bool {
self.data_available
}
pub fn is_bad(&self) -> bool { pub fn is_bad(&self) -> bool {
self.bad self.bad
} }