Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit65920b8

Browse files
Merge pull request#1229 from neondatabase/fix-connection-closure
Improve is_closed effectiveness
2 parentsa7a49a9 +e4be50d commit65920b8

File tree

6 files changed

+69
-8
lines changed

6 files changed

+69
-8
lines changed

‎postgres/src/client.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -487,6 +487,11 @@ impl Client {
487487
self.connection.block_on(self.client.batch_execute(query))
488488
}
489489

490+
/// Check the connection is alive and wait for the confirmation.
491+
pubfncheck_connection(&mutself) ->Result<(),Error>{
492+
self.connection.block_on(self.client.check_connection())
493+
}
494+
490495
/// Begins a new database transaction.
491496
///
492497
/// The transaction will roll back by default - use the `commit` method to commit it.

‎postgres/src/test.rs‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -508,3 +508,24 @@ fn check_send() {
508508
is_send::<Statement>();
509509
is_send::<Transaction<'_>>();
510510
}
511+
512+
#[test]
513+
fnis_closed(){
514+
letmut client =Client::connect("host=localhost port=5433 user=postgres",NoTls).unwrap();
515+
assert!(!client.is_closed());
516+
client.check_connection().unwrap();
517+
518+
let row = client.query_one("select pg_backend_pid()",&[]).unwrap();
519+
let pid:i32 = row.get(0);
520+
521+
{
522+
letmut client2 =Client::connect("host=localhost port=5433 user=postgres",NoTls).unwrap();
523+
client2
524+
.query("SELECT pg_terminate_backend($1)",&[&pid])
525+
.unwrap();
526+
}
527+
528+
assert!(!client.is_closed());
529+
client.check_connection().unwrap_err();
530+
assert!(client.is_closed());
531+
}

‎tokio-postgres/src/client.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -533,6 +533,12 @@ impl Client {
533533
simple_query::batch_execute(self.inner(), query).await
534534
}
535535

536+
/// Check the connection is alive and wait for the confirmation.
537+
pubasyncfncheck_connection(&self) ->Result<(),Error>{
538+
// sync is a very quick message to test the connection health.
539+
query::sync(self.inner()).await
540+
}
541+
536542
/// Begins a new database transaction.
537543
///
538544
/// The transaction will roll back by default - use the `commit` method to commit it.

‎tokio-postgres/src/connection.rs‎

Lines changed: 21 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -298,14 +298,7 @@ where
298298
self.parameters.get(name).map(|s|&**s)
299299
}
300300

301-
/// Polls for asynchronous messages from the server.
302-
///
303-
/// The server can send notices as well as notifications asynchronously to the client. Applications that wish to
304-
/// examine those messages should use this method to drive the connection rather than its `Future` implementation.
305-
///
306-
/// Return values of `None` or `Some(Err(_))` are "terminal"; callers should not invoke this method again after
307-
/// receiving one of those values.
308-
pubfnpoll_message(
301+
fnpoll_message_inner(
309302
&mutself,
310303
cx:&mutContext<'_>,
311304
) ->Poll<Option<Result<AsyncMessage,Error>>>{
@@ -323,6 +316,26 @@ where
323316
},
324317
}
325318
}
319+
320+
/// Polls for asynchronous messages from the server.
321+
///
322+
/// The server can send notices as well as notifications asynchronously to the client. Applications that wish to
323+
/// examine those messages should use this method to drive the connection rather than its `Future` implementation.
324+
///
325+
/// Return values of `None` or `Some(Err(_))` are "terminal"; callers should not invoke this method again after
326+
/// receiving one of those values.
327+
pubfnpoll_message(
328+
&mutself,
329+
cx:&mutContext<'_>,
330+
) ->Poll<Option<Result<AsyncMessage,Error>>>{
331+
matchself.poll_message_inner(cx){
332+
nominal @(Poll::Pending |Poll::Ready(Some(Ok(_)))) => nominal,
333+
terminal @(Poll::Ready(None) |Poll::Ready(Some(Err(_)))) =>{
334+
self.receiver.close();
335+
terminal
336+
}
337+
}
338+
}
326339
}
327340

328341
impl<S,T>FutureforConnection<S,T>

‎tokio-postgres/src/query.rs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -317,3 +317,13 @@ impl RowStream {
317317
self.rows_affected
318318
}
319319
}
320+
321+
pubasyncfnsync(client:&InnerClient) ->Result<(),Error>{
322+
let buf =Bytes::from_static(b"S\0\0\0\x04");
323+
letmut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
324+
325+
match responses.next().await?{
326+
Message::ReadyForQuery(_) =>Ok(()),
327+
_ =>Err(Error::unexpected_message()),
328+
}
329+
}

‎tokio-postgres/tests/test/main.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,12 @@ async fn scram_password_ok() {
146146
connect("user=scram_user password=password dbname=postgres").await;
147147
}
148148

149+
#[tokio::test]
150+
asyncfnsync(){
151+
let client =connect("user=postgres").await;
152+
client.check_connection().await.unwrap();
153+
}
154+
149155
#[tokio::test]
150156
asyncfnpipelined_prepare(){
151157
let client =connect("user=postgres").await;

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp