vertx-sql-client
vertx-sql-client copied to clipboard
PgSubscriberImpl.fetch() not implemented
Questions
The function PgSubscriberImpl.fetch() is not implemented and throws an UnsupportedOperationException
. This is a problem when using the Mutiny wrapper for the vertx-pg-client e.g. in Quarkus.
When you use PgSubscriber mechanism in a quarkus project leveraging the Mutiny framework, the following code runs into the UnsupportedOperationException
.
import javax.enterprise.context.ApplicationScoped
import io.quarkus.runtime.ShutdownEvent
import javax.enterprise.event.Observes
import io.quarkus.runtime.StartupEvent
import io.vertx.mutiny.core.Vertx
import io.vertx.mutiny.pgclient.pubsub.PgSubscriber
import io.vertx.pgclient.PgConnectOptions
import java.time.Duration
@ApplicationScoped
class PgChannelObserver {
fun onStart(@Observes ev: StartupEvent?, vertx: Vertx) {
val subscriber = PgSubscriber.subscriber(vertx, connectOptions())
subscriber.connect().await().atMost(Duration.ofSeconds(60))
val channel = subscriber.channel("testchannel") // I trigger the pg_notify directly in an pgsql session for that channel
channel.toMulti().subscribe().with { n -> print(n) }
}
private fun connectOptions() = PgConnectOptions()
.setPort(5432)
.setHost("localhost")
.setDatabase("rainbow_database")
.setUser("unicorn_user")
.setPassword("magical_password")
}
When I don't use Mutiny and instead use:
channel.handler { notif -> print("Notification received: $notif") }
everything works as expected.
I was wondering why the implementation for fetch is left out and if it is a Mutiny issue (needs to be addresses in smallrye) or related to an incomplete implementation of the PgSubscriber interface?
it's not implemented in vertx:
public ReadStream<String> fetch(long amount) {
throw new UnsupportedOperationException();
}
I think it could be implemented easily given that this class does not (purposely) buffers events
are you interested in contributing this @andreas-eberle