Add CheckpointingFileDataSupplier - can lock DSL values

api/build.gradle.kts

@@ -3,6 +3,7 @@ val kotlinVersion: String by project
 
 dependencies {
   implementation(kotlin("stdlib", kotlinVersion))
+  implementation(kotlin("reflect", kotlinVersion))
 
   implementation("org.jetbrains.kotlin:kotlin-stdlib")
   implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinxCoroutinesVersion")

api/src/main/kotlin/net/devslash/AcceptCallContext.kt

@@ -0,0 +1,5 @@
+package net.devslash
+
+interface AcceptCallContext<T> {
+  fun inject(): CallBuilder<T>.() -> Unit
+}

api/src/main/kotlin/net/devslash/Definitions.kt

@@ -35,7 +35,7 @@ data class Call<T>(
   val afterHooks: List<AfterHook>
 )
 
-interface RequestDataSupplier<T> {
+interface RequestDataSupplier<in T> {
   /**
    * Request data should be a closure that is safe to call on a per-request basis
    */

api/src/main/kotlin/net/devslash/FetchDsl.kt

@@ -3,6 +3,7 @@ package net.devslash
 import net.devslash.err.RetryOnTransitiveError
 import java.time.Duration
 import java.util.*
+import kotlin.reflect.KMutableProperty0
 
 /**
  * Version contains the current version defined in the build.gradle root file.
@@ -43,12 +44,12 @@ enum class HttpMethod {
 @FetchDSL
 @Suppress("MemberVisibilityCanBePrivate")
 open class CallBuilder<T>(private val url: String) {
-  var urlProvider: URLProvider? = null
-  var data: RequestDataSupplier<T>? = null
-  var body: HttpBody? = null
-  var type: HttpMethod = HttpMethod.GET
+  var urlProvider: URLProvider? by LockableValue(null)
+  var data: RequestDataSupplier<T>? by LockableValue(null)
+  var body: HttpBody? by LockableValue(null)
+  var type: HttpMethod by LockableValue(HttpMethod.GET)
   var headers: Map<String, List<Any>> = mapOf()
-  var onError: OnError? = RetryOnTransitiveError()
+  var onError: OnError? by LockableValue(RetryOnTransitiveError())
 
   private var preHooksList = mutableListOf<BeforeHook>()
   private var postHooksList = mutableListOf<AfterHook>()
@@ -65,6 +66,10 @@ open class CallBuilder<T>(private val url: String) {
     body = BodyBuilder().apply(block).build()
   }
 
+  fun inject(x: AcceptCallContext<T>) {
+    this.apply(x.inject())
+  }
+
   private fun mapHeaders(map: Map<String, List<Any>>): Map<String, List<HeaderValue>> {
     return map.mapValues { entry ->
       entry.value.map { value ->
@@ -87,6 +92,14 @@ open class CallBuilder<T>(private val url: String) {
       preHooksList, postHooksList
     )
   }
+
+  fun <T> lock(field: KMutableProperty0<RequestDataSupplier<T>?>) {
+    when (val delegate = field.getDelegate()) {
+      is LockableValue<*, *> -> {
+        delegate.lock()
+      }
+    }
+  }
 }
 
 fun replaceString(changes: Map<String, String>, str: String): String {

api/src/main/kotlin/net/devslash/LockableValue.kt

@@ -0,0 +1,58 @@
+package net.devslash
+
+import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.properties.ReadWriteProperty
+import kotlin.reflect.KProperty
+
+annotation class DSLLockedValue
+
+/**
+ * Lockable value is a delegate pattern that allows for a marker annotation to be used to lock.
+ *
+ * Doesn't work on primitives. If you want it to work on a primitive, then wrap it in a data class for now
+ */
+class LockableValue<R, T>(private var curValue: T) : ReadWriteProperty<R, T> {
+  private val locked = AtomicBoolean(false)
+
+  override fun getValue(thisRef: R, property: KProperty<*>): T {
+    return curValue
+  }
+
+  override fun setValue(thisRef: R, property: KProperty<*>, value: T) {
+
+    val annotation = if (value != null) value!!::class.java.getAnnotation(DSLLockedValue::class.java) else null
+    if (annotation != null) {
+      // Then we have someone who wants to be protected. Therefore lock it down
+      // Unless it's already set, in which complain
+      if (locked.get() && curValue != null) {
+        throw AlreadySetException(property, curValue)
+      }
+      if (locked.compareAndSet(false, true)) {
+        curValue = value
+        return
+      } else {
+        throw AlreadySetException(property, curValue)
+      }
+    }
+
+    // If we've already locked. Also throw
+    if (locked.get()) {
+      throw AlreadySetException(property, curValue)
+    }
+
+    // Otherwise we can set
+    curValue = value
+  }
+
+  fun lock() {
+    if (!locked.compareAndSet(false, true)) {
+      throw AlreadySetException2()
+    }
+  }
+
+  class AlreadySetException2 :
+    RuntimeException()
+
+  class AlreadySetException(kProperty: KProperty<*>, value: Any?) :
+    RuntimeException("Property \"${kProperty.name}\" has already been set to ${value}. Cannot be set again")
+}

examples/src/main/kotlin/net/devslash/examples/PipeExample.kt

@@ -5,13 +5,11 @@ import io.ktor.response.*
 import io.ktor.routing.*
 import io.ktor.server.engine.*
 import io.ktor.server.netty.*
-import net.devslash.action
+import net.devslash.*
 import net.devslash.data.FileDataSupplier
 import net.devslash.data.ListDataSupplier
-import net.devslash.mustGet
 import net.devslash.outputs.WriteFile
 import net.devslash.pipes.ResettablePipe
-import net.devslash.runHttp
 import java.net.ServerSocket
 import java.nio.file.Files
 import java.util.concurrent.TimeUnit
@@ -59,6 +57,7 @@ fun main() {
       }
       call<Int>(address) {
         data = ListDataSupplier(listOf(1, 2, 3))
+        lock(this::data)
         before {
           action {
             println(data.mustGet<Int>())
@@ -73,3 +72,9 @@ fun main() {
     server.stop(10, 10, TimeUnit.MILLISECONDS)
   }
 }
+
+
+@DSLLockedValue
+class ProtectedDS<T>(val ds: RequestDataSupplier<T>) : RequestDataSupplier<T> {
+  override suspend fun getDataForRequest(): RequestData? = ds.getDataForRequest()
+}

service/src/main/kotlin/net/devslash/HttpSessionManager.kt

@@ -37,7 +37,7 @@ class HttpSessionManager(private val engine: Driver, private val session: Sessio
 
   override fun <T> call(call: Call<T>): Exception? = call(call, DefaultCookieJar())
 
-  override fun <T> call(call: Call<T>, jar: CookieJar): Exception? = runBlocking(Dispatchers.Default) {
+  override fun <T> call(call: Call<T>, jar: CookieJar): Exception? = runBlocking {
     val channel: Channel<Envelope<Contents>> = Channel(session.concurrency * 2)
     launch(Dispatchers.IO) { RequestProducer().produceHttp(this@HttpSessionManager, call, jar, channel) }
 

service/src/main/kotlin/net/devslash/data/CheckpointingFileDataSupplier.kt

@@ -29,13 +29,18 @@ import java.util.concurrent.atomic.AtomicInteger
  *
  * A checkpointing supplier is also not expected to work
  */
+@DSLLockedValue
 class CheckpointingFileDataSupplier(
   fileName: String, //
   checkpointName: String, //
   private val split: String = " ", //
   private val checkpointPredicate: CheckpointPredicate = defaultCheckpointPredicate
 ) :
-  RequestDataSupplier<List<String>>, FullDataAfterHook, AutoCloseable, OnErrorWithState {
+  RequestDataSupplier<List<String>>,
+  FullDataAfterHook,
+  AutoCloseable,
+  OnErrorWithState,
+  AcceptCallContext<List<String>> {
 
   class CheckpointException(message: String) : RuntimeException(message)
 
@@ -62,13 +67,11 @@ class CheckpointingFileDataSupplier(
     lines = sourceFile.readLines()
   }
 
-  fun inject(callBuilder: CallBuilder<List<String>>) {
-    callBuilder.apply {
-      data = this@CheckpointingFileDataSupplier
-      onError = this@CheckpointingFileDataSupplier
-      after {
-        +this@CheckpointingFileDataSupplier
-      }
+  override fun inject(): CallBuilder<List<String>>.() -> Unit = {
+    data = this@CheckpointingFileDataSupplier
+    onError = this@CheckpointingFileDataSupplier
+    after {
+      +this@CheckpointingFileDataSupplier
     }
   }
 

service/src/test/kotlin/net/devslash/LockableValueTest.kt

@@ -0,0 +1,52 @@
+package net.devslash
+
+import org.hamcrest.CoreMatchers.*
+import org.hamcrest.MatcherAssert.assertThat
+import org.junit.Assert.assertThrows
+import org.junit.Test
+
+internal class LockableValueTest {
+
+  // What i can do is have `lock` as a
+  @DSLLockedValue
+  data class Locked(val b: Boolean)
+  data class Unlocked(val b: Boolean)
+
+  @Test
+  fun testLockWillAcceptNullable() {
+    val x: Int? by LockableValue(null)
+    assertThat(x, `is`(nullValue()))
+  }
+
+  @Test
+  fun testLockSetOnAnnotated() {
+    var x: Locked? by LockableValue(null)
+    x = Locked(true)
+    assertThrows(LockableValue.AlreadySetException::class.java) {
+      x = Locked(false)
+    }
+    assertThat(x, equalTo(Locked(true)))
+  }
+
+  @Test
+  fun testNotLockedTillFirstAnnotatedValue() {
+    var x: Any? by LockableValue(null)
+    x = Unlocked(false)
+    x = Locked(true)
+    assertThrows(LockableValue.AlreadySetException::class.java) {
+      x = Unlocked(true)
+    }
+    assertThat(x, equalTo(Locked(true)))
+  }
+
+  @Test
+  fun testLockDoesNotSetOnUnannotated() {
+    var x: Unlocked? by LockableValue(null)
+    x = Unlocked(true)
+    x = Unlocked(false)
+    assertThat(x, equalTo(Unlocked(false)))
+
+    x = null
+    assertThat(x, `is`(nullValue()))
+  }
+}

service/src/test/kotlin/net/devslash/data/CheckpointingFileDataSupplierTest.kt

@@ -2,10 +2,7 @@ package net.devslash.data
 
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runBlockingTest
-import net.devslash.Call
-import net.devslash.CallBuilder
-import net.devslash.HttpResponse
-import net.devslash.mustGet
+import net.devslash.*
 import net.devslash.outputs.LogResponse
 import net.devslash.util.basicRequest
 import net.devslash.util.basicResponse
@@ -69,7 +66,7 @@ internal class CheckpointingFileDataSupplierTest {
     val call: Call<List<String>>
     supplier.use {
       call = CallBuilder<List<String>>("http://example.com").apply {
-        it.inject(this)
+        inject(supplier)
         after {
           // Test append, not overwrite
           +LogResponse()
@@ -96,6 +93,12 @@ internal class CheckpointingFileDataSupplierTest {
     val rd = supplier.getDataForRequest()!!
     val rd2 = supplier.getDataForRequest()!!
 
+    runHttp {
+      call("T") {
+        inject(supplier)
+      }
+    }
+
     // In this form, 200 is failure
     supplier.accept(basicRequest(), basicResponse(), rd)
     supplier.accept(basicRequest(), HttpResponse(URI(basicUrl), 404, mapOf(), ByteArray(0)), rd2)