KaneAI - World’s First E2E Software Testing Agent.Signup for Private Beta Join TestMu Conference

How to use onSubscribe method of org.testcontainers.r2dbc.ConnectionPublisher class

Best Testcontainers-java code snippet using org.testcontainers.r2dbc.ConnectionPublisher.onSubscribe

Source:ConnectionPublisher.java

...19        this.futureSupplier = futureSupplier;20    }21    @Override22    public void subscribe(Subscriber<? super Connection> actual) {23        actual.onSubscribe(new StateMachineSubscription(actual));24    }25    private class StateMachineSubscription implements Subscription {26        private final Subscriber<? super Connection> actual;27        Subscription subscriptionState;28        StateMachineSubscription(Subscriber<? super Connection> actual) {29            this.actual = actual;30            subscriptionState = new WaitRequestSubscriptionState();31        }32        @Override33        public synchronized void request(long n) {34            subscriptionState.request(n);35        }36        @Override37        public synchronized void cancel() {38            subscriptionState.cancel();39        }40        synchronized void transitionTo(SubscriptionState newState) {41            subscriptionState = newState;42            newState.enter();43        }44        abstract class SubscriptionState implements Subscription {45            void enter() {46            }47        }48        class WaitRequestSubscriptionState extends SubscriptionState {49            @Override50            public void request(long n) {51                transitionTo(new WaitFutureCompletionSubscriptionState());52            }53            @Override54            public void cancel() {55            }56        }57        class WaitFutureCompletionSubscriptionState extends SubscriptionState {58            private CompletableFuture<ConnectionFactory> future;59            @Override60            void enter() {61                this.future = futureSupplier.get();62                future.whenComplete((connectionFactory, e) -> {63                    if (e != null) {64                        actual.onSubscribe(EmptySubscription.INSTANCE);65                        actual.onError(e);66                        return;67                    }68                    Publisher<? extends Connection> publisher = connectionFactory.create();69                    transitionTo(new ProxySubscriptionState(publisher));70                });71            }72            @Override73            public void request(long n) {74            }75            @Override76            public void cancel() {77                future.cancel(true);78            }79        }80        class ProxySubscriptionState extends SubscriptionState implements Subscriber<Connection> {81            private final Publisher<? extends Connection> publisher;82            private Subscription s;83            private boolean cancelled = false;84            ProxySubscriptionState(Publisher<? extends Connection> publisher) {85                this.publisher = publisher;86            }87            @Override88            void enter() {89                publisher.subscribe(this);90            }91            @Override92            public void request(long n) {93                // Ignore94            }95            @Override96            public synchronized void cancel() {97                cancelled = true;98                if (s != null) {99                    s.cancel();100                }101            }102            @Override103            public synchronized void onSubscribe(Subscription s) {104                this.s = s;105                if (!cancelled) {106                    s.request(1);107                } else {108                    s.cancel();109                }110            }111            @Override112            public void onNext(Connection connection) {113                actual.onNext(connection);114            }115            @Override116            public void onError(Throwable t) {117                actual.onError(t);...

onSubscribe

Using AI Code Generation

1class ConnectionPublisher {2  def onSubscribe(subscription: Subscription) {3    subscription.request(1)4  }5}6class ConnectionPublisher {7  def onNext(connection: Connection) {8    subscription.request(1)9  }10}11class ConnectionPublisher {12  def onError(throwable: Throwable) {13    subscription.request(1)14  }15}16class ConnectionPublisher {17  def onComplete() {18    subscription.request(1)19  }20}21class ConnectionPublisher {22  def request(n: Long) {23    subscription.request(n)24  }25}26class ConnectionPublisher {27  def cancel() {28    subscription.cancel()29  }30}31class ConnectionPublisher {32  def get(): Connection = {33    if (throwable != null) {34    }35    if (connection == null) {36      throw new IllegalStateException("Connection is not available")37    }38  }39}40class ConnectionPublisher {41  def get(timeout: Long, timeUnit: TimeUnit): Connection = {42    if (throwable != null) {43    }44    if (connection == null) {45      throw new IllegalStateException("Connection is not available")46    }47  }48}49class ConnectionPublisher {50  def get(timeout: Duration): Connection = {51    if (throwable != null) {52    }53    if (connection == null) {54      throw new IllegalStateException("Connection is not available")55    }56  }57}58class ConnectionPublisher {59  def close() {60    if (connection != null) {61      connection.close()62    }63    subscription.cancel()64  }65}

onSubscribe

Using AI Code Generation

1public   class   TestContainersR2dbcTest   {2     public   void   testR2dbc ()   {3          try  ( PostgresContainer  container  =   new  PostgresContainer ()) {4              container . start ();5              String  jdbcUrl  =  container . getJdbcUrl ();6              String  r2dbcUrl  =  container . getR2dbcUrl ();7              ConnectionFactory  connectionFactory  =   new  PostgresqlConnectionFactory ( new  PostgresqlConnectionConfiguration . Builder ()8                     . host ( container . getHost ())9                     . port ( container . getPort ())10                     . database ( container . getDatabaseName ())11                     . username ( container . getUsername ())12                     . password ( container . getPassword ())13                     . build ());14              Mono .< Connection > create ( sink  ->  connectionFactory . create ())15                     . map ( connection  ->   {16                          Mono .< Statement > create ( sink2  ->  connection . createStatement ( " CREATE TABLE IF NOT EXISTS users ( id SERIAL, name VARCHAR ( 200 ) NOT NULL, PRIMARY KEY ( id ) ) " ))17                                 . map ( statement  ->   {18                                      statement . execute ()19                                             . as ( Result :: getRowsUpdated )20                                             . subscribe ( rowsUpdated  ->  sink2 . success ( statement ));21                                     return  statement ;22                                })23                                 . subscribe ();24                          Mono .< Statement > create ( sink2  ->  connection . createStatement ( " INSERT INTO users ( name ) VALUES ( ' Test ' ) " ))25                                 . map ( statement  ->   {26                                      statement . execute ()27                                             . as ( Result :: getRowsUpdated )28                                             . subscribe ( rowsUpdated  ->  sink2

onSubscribe

Using AI Code Generation

1public   class   TestContainersR2dbcTest   {2     public   void   testR2dbc ()   {3          try  ( PostgresContainer  container  =   new  PostgresContainer ()) {4              container . start ();5              String  jdbcUrl  =  container . getJdbcUrl ();6              String  r2dbcUrl  =  container . getR2dbcUrl ();7              ConnectionFactory  connectionFactory  =   new  PostgresqlConnectionFactory ( new  PostgresqlConnectionConfiguration . Builder ()8                     . host ( container . getHost ())9                     . port ( container . getPort ())10                     . database ( container . getDatabaseName ())11                     . username ( container . getUsername ())12                     . password ( container . getPassword ())13                     . build ());14              Mono .< Connection > create ( sink  ->  connectionFactory . create ())15                     . map ( connection  ->   {16                          Mono .< Statement > create ( sink2  ->  connection . createStatement ( " CREATE TABLE IF NOT EXISTS users ( id SERIAL, name VARCHAR ( 200 ) NOT NULL, PRIMARY KEY ( id ) ) " ))17                                 . map ( statement  ->   {18                                      statement . execute ()19                                             . as ( Result :: getRowsUpdated )20                                             . subscribe ( rowsUpdated  ->  sink2 . success ( statement ));21                                     return  statement ;22                                })23                                 . subscribe ();24                          Mono .< Statement > create ( sink2  ->  connection . createStatement ( " INSERT INTO users ( name ) VALUES ( ' Test ' ) " ))25                                 . map ( statement  ->   {26                                      statement . execute ()27                                             . as ( Result :: getRowsUpdated )28                                             . subscribe ( rowsUpdated  ->  sink2

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.