gradle sucks etc

core-lib/build.gradle.kts

@@ -8,7 +8,6 @@ plugins {
 }
 
 group = "de.kentoj.scrow"
-version = "0.4"
 
 repositories {
     mavenCentral()

core-lib/src/main/java/de/kentoj/scrowlib/connection/DatabaseConnectionFactory.java

@@ -0,0 +1,8 @@
+package de.kentoj.scrowlib.connection;
+
+import org.springframework.r2dbc.core.DatabaseClient;
+
+public interface DatabaseConnectionFactory extends AutoCloseable {
+
+    DatabaseClient create();
+}

core-lib/src/main/java/de/kentoj/scrowlib/connection/DatabaseConnectionFactoryImpl.java

@@ -0,0 +1,31 @@
+package de.kentoj.scrowlib.connection;
+
+import io.r2dbc.pool.ConnectionPool;
+import io.r2dbc.pool.ConnectionPoolConfiguration;
+import io.r2dbc.spi.ConnectionFactory;
+import lombok.RequiredArgsConstructor;
+import org.springframework.r2dbc.core.DatabaseClient;
+
+@RequiredArgsConstructor
+public class DatabaseConnectionFactoryImpl implements DatabaseConnectionFactory {
+
+    private final ConnectionPool backingFactory;
+
+    public DatabaseConnectionFactoryImpl(ConnectionFactory connectionFactory) {
+        this(new ConnectionPool(ConnectionPoolConfiguration.builder()
+                .connectionFactory(connectionFactory)
+                .build()));
+    }
+
+    // FIXME
+    // its been a while and i no longer remember what had to be fixed
+    @Override
+    public DatabaseClient create() {
+        return DatabaseClient.create(backingFactory);
+    }
+
+    @Override
+    public void close() {
+        backingFactory.close();
+    }
+}

core-lib/src/main/java/de/kentoj/scrowlib/messaging/NatsRepository.java

@@ -6,9 +6,12 @@ import io.nats.client.Dispatcher;
 import io.nats.client.Message;
 import org.bson.Document;
 import org.jetbrains.annotations.NotNull;
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 public class NatsRepository {
 
     private final Connection nats;
@@ -34,15 +37,13 @@ public class NatsRepository {
 
     public Flux<@NotNull SubscriptionContext> subscribe(String subject) {
         return Flux.create(sink -> {
-            var handler = dispatcher.subscribe(subject, msg -> {
+            dispatcher.subscribe(subject, msg -> {
                 try {
                     sink.next(new SubscriptionContext(msg));
                 } catch (Throwable e) {
                     sink.error(e);
                 }
             });
-            sink.onCancel(handler::unsubscribe);
-            sink.onDispose(handler::unsubscribe);
         });
     }
 

core-lib/src/main/java/de/kentoj/scrowlib/messaging/SubscriptionContext.java

@@ -1,6 +1,5 @@
 package de.kentoj.scrowlib.messaging;
 
-import de.kentoj.scrowlib.messaging.respose.SafeResult;
 import io.nats.client.Message;
 import lombok.AllArgsConstructor;
 import lombok.Value;
@@ -13,6 +12,6 @@ public class SubscriptionContext {
     Document document;
 
     public SubscriptionContext(Message msg) {
-        this(msg, SafeResult.fromMessage(msg).unwrap());
+        this(msg, DocumentRepository.fromMessage(msg));
     }
 }

core-lib/src/main/java/de/kentoj/scrowlib/messaging/respose/SafeResult.java

@@ -3,6 +3,8 @@ package de.kentoj.scrowlib.messaging.respose;
 import de.kentoj.scrowlib.messaging.DocumentRepository;
 import io.nats.client.Message;
 import org.bson.Document;
+import org.jetbrains.annotations.NotNull;
+import reactor.core.publisher.Mono;
 
 public abstract sealed class SafeResult permits SafeErrorResult, SafeSuccessResult {
 
@@ -61,4 +63,9 @@ public abstract sealed class SafeResult permits SafeErrorResult, SafeSuccessResu
         }
         throw new IllegalArgumentException("document does not conform to SafeDocument format");
     }
+
+    public static Mono<@NotNull SafeResult> mono(Mono<@NotNull Document> mono) {
+        return mono.map(SafeResult::wrap)
+                .onErrorResume(err -> Mono.just(SafeResult.wrap(err)));
+    }
 }