From d6df59d8d5ba9134056584d8502920ac0a94d468 Mon Sep 17 00:00:00 2001
From: Andrea Vinci <andrewinci@gmail.com>
Date: Sun, 15 Nov 2020 09:52:39 +0000
Subject: [PATCH] Add refresh button (#138)

* Cache list topic in cluster scope

* Add refresh button

* Add test refresh button in topic list

* Add test for refresh schema registry list

* Fix PR lint

* Add unit test for register schema

* Remove code duplication

* Rename buttons

* Fix text in button alignment

Co-authored-by: Auto Lint <autolint@lint.local>
---
 .../integrationtest/RefreshButtonTests.kt     | 64 +++++++++++++++++++
 .../helpers/IntegrationTestFixture.kt         | 35 ++++++++--
 .../kotlin/insulator/ui/component/AppBar.kt   | 28 ++++++--
 .../kotlin/insulator/ui/style/AppBarStyle.kt  |  6 +-
 .../main/topic/ListTopicViewModel.kt          |  2 +
 .../kotlin/insulator/views/main/MainView.kt   |  7 +-
 .../main/schemaregistry/ListSchemaView.kt     | 23 ++++---
 .../views/main/schemaregistry/SchemaView.kt   |  9 +--
 .../views/main/topic/ListTopicView.kt         | 23 +++++--
 .../views/main/topic/ProducerView.kt          |  3 +-
 .../insulator/views/main/topic/TopicView.kt   | 12 ++--
 .../kotlin/insulator/kafka/SchemaRegistry.kt  |  5 ++
 .../insulator/kafka/SchemaRegistryTest.kt     | 28 ++++++++
 13 files changed, 198 insertions(+), 47 deletions(-)
 create mode 100644 app/src/integrationTest/kotlin/insulator/integrationtest/RefreshButtonTests.kt

diff --git a/app/src/integrationTest/kotlin/insulator/integrationtest/RefreshButtonTests.kt b/app/src/integrationTest/kotlin/insulator/integrationtest/RefreshButtonTests.kt
new file mode 100644
index 00000000..4ed3b6fe
--- /dev/null
+++ b/app/src/integrationTest/kotlin/insulator/integrationtest/RefreshButtonTests.kt
@@ -0,0 +1,64 @@
+package insulator.integrationtest
+
+import insulator.integrationtest.helpers.IntegrationTestFixture
+import insulator.integrationtest.helpers.click
+import insulator.integrationtest.helpers.lookupFirst
+import insulator.integrationtest.helpers.screenShoot
+import insulator.integrationtest.helpers.selectCluster
+import insulator.integrationtest.helpers.waitWindowWithTitle
+import io.kotest.core.spec.style.FreeSpec
+import io.kotest.matchers.shouldBe
+import javafx.scene.Node
+import javafx.scene.control.Button
+import javafx.scene.control.Label
+import javafx.scene.control.ListView
+import kotlinx.coroutines.delay
+import tornadofx.CssRule
+import tornadofx.Stylesheet.Companion.listView
+import kotlin.time.ExperimentalTime
+
+@ExperimentalTime
+class RefreshButtonTests : FreeSpec({
+
+    "Test refresh topic button" - {
+        IntegrationTestFixture().use { fixture ->
+            fixture.startAppWithKafkaCuster("Test cluster")
+            selectCluster(fixture.currentKafkaCluster)
+
+            val mainView = waitWindowWithTitle("Insulator")
+            suspend fun getListViewItems() = mainView.lookupFirst<ListView<String>>(listView).items
+
+            "Refresh topic list" {
+                val topicName = "test-new-topic"
+                // create topic
+                fixture.createTopic(topicName)
+                delay(1_000)
+                // the topic shouldn't be visible
+                getListViewItems().count { it == topicName } shouldBe 0
+                // click the refresh button wil load the new topic
+                mainView.lookupFirst<Button>(CssRule.id("button-refresh")).click()
+                screenShoot("refresh-topic-list")
+                // the list of topic is now updated
+                getListViewItems().count { it == topicName } shouldBe 1
+                mainView.lookupFirst<Label>(CssRule.id("topic-$topicName")).text shouldBe topicName
+            }
+
+            "Refresh schema list" {
+                val schemaName = "test-new-topic-schema"
+                // select schema registry
+                mainView.lookupFirst<Node>(CssRule.id("sidebar-item-schema-registry")).click()
+                // create the schema
+                fixture.createTestSchema(schemaName)
+                delay(1_000)
+                // the topic shouldn't be visible
+                getListViewItems().count { it == schemaName } shouldBe 0
+                // click the refresh button wil load the new topic
+                mainView.lookupFirst<Button>(CssRule.id("button-refresh")).click()
+                screenShoot("refresh-schema-registry-list")
+                // the list of topic is now updated
+                getListViewItems().count { it == schemaName } shouldBe 1
+                mainView.lookupFirst<Label>(CssRule.id("schema-$schemaName")).text shouldBe schemaName
+            }
+        }
+    }
+})
diff --git a/app/src/integrationTest/kotlin/insulator/integrationtest/helpers/IntegrationTestFixture.kt b/app/src/integrationTest/kotlin/insulator/integrationtest/helpers/IntegrationTestFixture.kt
index 3730ab1e..c8f869eb 100644
--- a/app/src/integrationTest/kotlin/insulator/integrationtest/helpers/IntegrationTestFixture.kt
+++ b/app/src/integrationTest/kotlin/insulator/integrationtest/helpers/IntegrationTestFixture.kt
@@ -3,6 +3,7 @@ package insulator.integrationtest.helpers
 import insulator.Insulator
 import insulator.configuration.ConfigurationRepo
 import insulator.kafka.AdminApi
+import insulator.kafka.SchemaRegistry
 import insulator.kafka.adminApi
 import insulator.kafka.local.SchemaRegistryContainer
 import insulator.kafka.model.Cluster
@@ -10,6 +11,7 @@ import insulator.kafka.model.SchemaRegistryConfiguration
 import insulator.kafka.model.Topic
 import insulator.kafka.producer.Producer
 import insulator.kafka.producer.stringProducer
+import insulator.kafka.schemaRegistry
 import insulator.test.helper.deleteTestSandboxFolder
 import insulator.test.helper.getTestSandboxFolder
 import org.testcontainers.containers.KafkaContainer
@@ -20,10 +22,11 @@ import java.io.Closeable
 
 class IntegrationTestFixture : Closeable {
     private lateinit var adminApi: AdminApi
+    private lateinit var schemaRegistry: SchemaRegistry
     lateinit var stringProducer: Producer
     private val currentHomeFolder = getTestSandboxFolder().toAbsolutePath()
     private val kafka = KafkaContainer()
-    private val schemaRegistry = SchemaRegistryContainer().withKafka(kafka)
+    private val schemaRegistryContainer = SchemaRegistryContainer().withKafka(kafka)
     lateinit var currentKafkaCluster: Cluster
 
     init {
@@ -40,12 +43,13 @@ class IntegrationTestFixture : Closeable {
             name = clusterName,
             endpoint = kafka.bootstrapServers,
             schemaRegistryConfig = if (createSchemaRegistry) {
-                schemaRegistry.start()
-                schemaRegistry.waitingFor(Wait.forListeningPort())
-                SchemaRegistryConfiguration(schemaRegistry.endpoint)
+                schemaRegistryContainer.start()
+                schemaRegistryContainer.waitingFor(Wait.forListeningPort())
+                SchemaRegistryConfiguration(schemaRegistryContainer.endpoint)
             } else SchemaRegistryConfiguration()
         )
         startApp(currentKafkaCluster)
+        if (createSchemaRegistry) schemaRegistry = schemaRegistry(currentKafkaCluster)
         adminApi = adminApi(currentKafkaCluster)
         stringProducer = stringProducer(currentKafkaCluster)
     }
@@ -68,8 +72,29 @@ class IntegrationTestFixture : Closeable {
         kotlin.runCatching { FxToolkit.cleanupApplication(FX.application) }
         deleteTestSandboxFolder()
         kafka.runCatching { stop(); close() }
-        schemaRegistry.runCatching { stop(); close() }
+        schemaRegistryContainer.runCatching { stop(); close() }
     }
 
     suspend fun createTopic(s: String) = adminApi.createTopics(Topic(s, partitionCount = 3, replicationFactor = 1))
+    fun createTestSchema(schemaName: String) = schemaRegistry.register(schemaName, testSchema)
+
+    val testSchema =
+        """
+        {
+          "type": "record", 
+          "name": "value_test_schema", 
+          "namespace": "com.mycorp.mynamespace", 
+          "doc": "Sample schema to help you get started.", 
+          "fields": [{
+              "name": "test", 
+              "type": ["null", {
+                  "type": "bytes", 
+                  "logicalType": "decimal", 
+                  "precision": 5, 
+                  "scale": 2
+                }], 
+              "default": null
+            }]
+        }
+        """.trimIndent()
 }
diff --git a/app/src/main/kotlin/insulator/ui/component/AppBar.kt b/app/src/main/kotlin/insulator/ui/component/AppBar.kt
index ec08b03a..4155b855 100644
--- a/app/src/main/kotlin/insulator/ui/component/AppBar.kt
+++ b/app/src/main/kotlin/insulator/ui/component/AppBar.kt
@@ -1,16 +1,32 @@
 package insulator.ui.component
 
 import insulator.ui.style.AppBarStyle
+import javafx.beans.value.ObservableStringValue
 import javafx.event.EventTarget
 import javafx.geometry.Pos
-import javafx.scene.layout.Priority
+import javafx.scene.control.Button
 import tornadofx.addClass
-import tornadofx.hgrow
+import tornadofx.borderpane
+import tornadofx.buttonbar
 import tornadofx.vbox
 
-fun EventTarget.appBar(op: EventTarget.() -> Unit) =
-    vbox(alignment = Pos.TOP_LEFT) {
-        op()
-        hgrow = Priority.ALWAYS
+data class AppBarBuilder(
+    var title: String = "",
+    var subtitle: ObservableStringValue? = null,
+    var buttons: List<Button> = emptyList()
+)
+
+fun EventTarget.appBar(op: AppBarBuilder.() -> Unit) {
+    val builder = AppBarBuilder().apply(op)
+    this.borderpane {
+        center = vbox {
+            h1(builder.title)
+            builder.subtitle?.let { subTitle(builder.subtitle!!) }
+            alignment = Pos.CENTER_LEFT
+        }
+        right = buttonbar {
+            buttons.addAll(builder.buttons)
+        }
         addClass(AppBarStyle.appBar)
     }
+}
diff --git a/app/src/main/kotlin/insulator/ui/style/AppBarStyle.kt b/app/src/main/kotlin/insulator/ui/style/AppBarStyle.kt
index d77146b8..bf697fcf 100644
--- a/app/src/main/kotlin/insulator/ui/style/AppBarStyle.kt
+++ b/app/src/main/kotlin/insulator/ui/style/AppBarStyle.kt
@@ -19,12 +19,14 @@ class AppBarStyle : Stylesheet() {
         appBar {
             translateY = -theme.viewPadding
             borderInsets = multi(box(-theme.viewPadding, -theme.viewPadding, 0.px, -theme.viewPadding))
-            spacing = 5.0.px
-            alignment = Pos.CENTER_LEFT
             borderColor = multi(box(theme.backgroundColor, theme.backgroundColor, theme.lightGray, theme.backgroundColor))
             minHeight = 5.em
         }
 
+        button {
+            alignment = Pos.CENTER
+        }
+
         burgerButton {
             padding = box(1.em)
             textFill = theme.mainColor
diff --git a/app/src/main/kotlin/insulator/viewmodel/main/topic/ListTopicViewModel.kt b/app/src/main/kotlin/insulator/viewmodel/main/topic/ListTopicViewModel.kt
index 00b25f2c..9bdb6120 100644
--- a/app/src/main/kotlin/insulator/viewmodel/main/topic/ListTopicViewModel.kt
+++ b/app/src/main/kotlin/insulator/viewmodel/main/topic/ListTopicViewModel.kt
@@ -1,5 +1,6 @@
 package insulator.viewmodel.main.topic
 
+import insulator.di.ClusterScope
 import insulator.di.factories.TopicComponentFactory
 import insulator.helper.dispatch
 import insulator.helper.runOnFXThread
@@ -19,6 +20,7 @@ import tornadofx.SortedFilteredList
 import tornadofx.whenUndockedOnce
 import javax.inject.Inject
 
+@ClusterScope
 class ListTopicViewModel @Inject constructor(
     val cluster: Cluster,
     val adminApi: AdminApi,
diff --git a/app/src/main/kotlin/insulator/views/main/MainView.kt b/app/src/main/kotlin/insulator/views/main/MainView.kt
index a4dfc2f9..85b9b9bd 100644
--- a/app/src/main/kotlin/insulator/views/main/MainView.kt
+++ b/app/src/main/kotlin/insulator/views/main/MainView.kt
@@ -94,8 +94,8 @@ class MainView @Inject constructor(
                 button("Info") { action { clusterView.show(false) }; id = "button-cluster-info" }
             }
             center = vbox {
-                menuItem("Topics", ICON_TOPICS) { viewModel.setContentList(ListTopicView::class) }
-                menuItem("Schema Registry", ICON_REGISTRY) { viewModel.setContentList(ListSchemaView::class) }
+                menuItem("Topics", ICON_TOPICS, "sidebar-item-topics") { viewModel.setContentList(ListTopicView::class) }
+                menuItem("Schema Registry", ICON_REGISTRY, "sidebar-item-schema-registry") { viewModel.setContentList(ListSchemaView::class) }
             }
             bottom = hbox(alignment = Pos.CENTER) {
                 themeButton { themeHelper.dispatch { changeTheme() } }
@@ -105,12 +105,13 @@ class MainView @Inject constructor(
             maxWidth = SIDEBAR_WIDTH
         }
 
-    private fun EventTarget.menuItem(name: String, icon: String, onClick: () -> Unit) =
+    private fun EventTarget.menuItem(name: String, icon: String, id: String, onClick: () -> Unit) =
         hbox(spacing = 5.0) {
             imageview(Image(icon)) { fitHeight = 35.0; fitWidth = 35.0; }
             h2(name)
             onMouseClicked = EventHandler { onClick() }
             addClass(MainViewStyle.sidebarItem)
+            this.id = id
         }
 
     private fun setSize() {
diff --git a/app/src/main/kotlin/insulator/views/main/schemaregistry/ListSchemaView.kt b/app/src/main/kotlin/insulator/views/main/schemaregistry/ListSchemaView.kt
index 3a98f79a..5025734c 100644
--- a/app/src/main/kotlin/insulator/views/main/schemaregistry/ListSchemaView.kt
+++ b/app/src/main/kotlin/insulator/views/main/schemaregistry/ListSchemaView.kt
@@ -4,16 +4,17 @@ import insulator.di.ClusterScope
 import insulator.helper.dispatch
 import insulator.ui.common.InsulatorView
 import insulator.ui.component.appBar
-import insulator.ui.component.h1
 import insulator.ui.component.searchBox
-import insulator.ui.component.subTitle
+import insulator.ui.style.ButtonStyle
 import insulator.viewmodel.main.schemaregistry.ListSchemaViewModel
 import insulator.viewmodel.main.schemaregistry.LoadSchemaListError
 import javafx.event.EventTarget
 import javafx.scene.control.SelectionMode
 import javafx.scene.layout.Priority
+import tornadofx.action
+import tornadofx.addClass
 import tornadofx.bindSelected
-import tornadofx.hbox
+import tornadofx.button
 import tornadofx.label
 import tornadofx.listview
 import tornadofx.onDoubleClick
@@ -28,10 +29,9 @@ class ListSchemaView @Inject constructor(
 
     override val root = vbox(spacing = 5.0) {
         appBar {
-            hbox {
-                h1("Schema registry")
-            }
-            subTitle(viewModel.subtitleProperty)
+            title = "Schema registry"
+            subtitle = viewModel.subtitleProperty
+            buttons = listOf(refreshButton())
         }
         searchBox(viewModel.searchItemProperty, currentView = this@ListSchemaView)
         schemasListView()
@@ -39,7 +39,7 @@ class ListSchemaView @Inject constructor(
 
     private fun EventTarget.schemasListView() =
         listview<String> {
-            cellFormat { graphic = label(it) }
+            cellFormat { graphic = label(it) { id = "schema-$it" } }
             onDoubleClick { viewModel.dispatch { showSchema() } }
             itemsProperty().set(viewModel.filteredSchemasProperty)
             bindSelected(viewModel.selectedSchemaProperty)
@@ -55,4 +55,11 @@ class ListSchemaView @Inject constructor(
             else -> viewModel.refresh()
         }
     }
+
+    private fun EventTarget.refreshButton() =
+        button("Refresh") {
+            id = "button-refresh"
+            action { dispatch { viewModel.refresh() } }
+            addClass(ButtonStyle.blueButton)
+        }
 }
diff --git a/app/src/main/kotlin/insulator/views/main/schemaregistry/SchemaView.kt b/app/src/main/kotlin/insulator/views/main/schemaregistry/SchemaView.kt
index b13a49cf..55ce5502 100644
--- a/app/src/main/kotlin/insulator/views/main/schemaregistry/SchemaView.kt
+++ b/app/src/main/kotlin/insulator/views/main/schemaregistry/SchemaView.kt
@@ -7,7 +7,6 @@ import insulator.ui.common.InsulatorTabView
 import insulator.ui.component.appBar
 import insulator.ui.component.confirmationButton
 import insulator.ui.component.fieldName
-import insulator.ui.component.h1
 import insulator.ui.style.theme
 import insulator.viewmodel.main.schemaregistry.SchemaViewModel
 import javafx.event.EventTarget
@@ -36,10 +35,8 @@ class SchemaView @Inject constructor(
 
     override val root = vbox {
         appBar {
-            hbox(alignment = Pos.CENTER_LEFT, spacing = 5.0) {
-                h1(viewModel.nameProperty.value)
-                deleteButton()
-            }
+            title = viewModel.nameProperty.value
+            buttons = listOf(deleteButton())
         }
         hbox(alignment = Pos.CENTER_LEFT) {
             fieldName("Schema")
@@ -59,7 +56,7 @@ class SchemaView @Inject constructor(
         }
 
     private fun EventTarget.deleteButton() =
-        confirmationButton("delete", "The schema \"${viewModel.nameProperty.value}\" will be removed.") {
+        confirmationButton("Delete", "The schema \"${viewModel.nameProperty.value}\" will be removed.") {
             viewModel.delete()
             closeTab()
         }
diff --git a/app/src/main/kotlin/insulator/views/main/topic/ListTopicView.kt b/app/src/main/kotlin/insulator/views/main/topic/ListTopicView.kt
index d06081ae..fd9ed705 100644
--- a/app/src/main/kotlin/insulator/views/main/topic/ListTopicView.kt
+++ b/app/src/main/kotlin/insulator/views/main/topic/ListTopicView.kt
@@ -1,36 +1,37 @@
 package insulator.views.main.topic
 
+import insulator.di.ClusterScope
 import insulator.helper.dispatch
 import insulator.ui.common.InsulatorView
 import insulator.ui.component.action
 import insulator.ui.component.appBar
-import insulator.ui.component.h1
 import insulator.ui.component.searchBox
-import insulator.ui.component.subTitle
+import insulator.ui.style.ButtonStyle
 import insulator.viewmodel.main.topic.ListTopicViewModel
 import insulator.views.configurations.ListClusterView
 import javafx.event.EventTarget
+import javafx.geometry.Pos
 import javafx.scene.control.SelectionMode
 import javafx.scene.layout.Priority
 import tornadofx.action
+import tornadofx.addClass
 import tornadofx.bindSelected
 import tornadofx.borderpane
 import tornadofx.button
-import tornadofx.hbox
 import tornadofx.label
 import tornadofx.listview
 import tornadofx.vbox
 import tornadofx.vgrow
 import javax.inject.Inject
 
+@ClusterScope
 class ListTopicView @Inject constructor(override val viewModel: ListTopicViewModel) : InsulatorView("Topics") {
 
     override val root = vbox(spacing = 5.0) {
         appBar {
-            hbox {
-                h1("Topics")
-            }
-            subTitle(viewModel.subtitleProperty)
+            title = "Topics"
+            subtitle = viewModel.subtitleProperty
+            buttons = listOf(refreshButton())
         }
         borderpane {
             left = createTopicButton()
@@ -57,6 +58,14 @@ class ListTopicView @Inject constructor(override val viewModel: ListTopicViewMod
             id = "button-create-topic"
         }
 
+    private fun EventTarget.refreshButton() =
+        button("Refresh") {
+            action { viewModel.dispatch { viewModel.refresh() } }
+            addClass(ButtonStyle.blueButton)
+            alignment = Pos.CENTER
+            id = "button-refresh"
+        }
+
     override fun onError(throwable: Throwable) {
         replaceWith<ListClusterView>()
     }
diff --git a/app/src/main/kotlin/insulator/views/main/topic/ProducerView.kt b/app/src/main/kotlin/insulator/views/main/topic/ProducerView.kt
index 48d04af6..edd50e5a 100644
--- a/app/src/main/kotlin/insulator/views/main/topic/ProducerView.kt
+++ b/app/src/main/kotlin/insulator/views/main/topic/ProducerView.kt
@@ -7,7 +7,6 @@ import insulator.kafka.consumer.DeserializationFormat
 import insulator.ui.common.InsulatorView
 import insulator.ui.component.appBar
 import insulator.ui.component.fieldName
-import insulator.ui.component.h1
 import insulator.viewmodel.main.topic.ProducerViewModel
 import javafx.beans.binding.Bindings
 import javafx.event.EventTarget
@@ -39,7 +38,7 @@ class ProducerView @Inject constructor(
     private val recordValueTextArea = TextArea()
 
     override val root = vbox(spacing = 10.0) {
-        appBar { h1(viewModel.topic.name) }
+        appBar { title = viewModel.topic.name }
         fieldName("Key")
         textfield(viewModel.keyProperty) { id = "field-producer-key" }
 
diff --git a/app/src/main/kotlin/insulator/views/main/topic/TopicView.kt b/app/src/main/kotlin/insulator/views/main/topic/TopicView.kt
index 1169f042..9a42baf1 100644
--- a/app/src/main/kotlin/insulator/views/main/topic/TopicView.kt
+++ b/app/src/main/kotlin/insulator/views/main/topic/TopicView.kt
@@ -10,9 +10,7 @@ import insulator.ui.common.InsulatorTabView
 import insulator.ui.component.appBar
 import insulator.ui.component.confirmationButton
 import insulator.ui.component.fieldName
-import insulator.ui.component.h1
 import insulator.ui.component.searchBox
-import insulator.ui.component.subTitle
 import insulator.ui.style.ButtonStyle
 import insulator.viewmodel.main.topic.RecordViewModel
 import insulator.viewmodel.main.topic.TopicViewModel
@@ -55,11 +53,9 @@ class TopicView @Inject constructor(
 
     override val root = vbox {
         appBar {
-            hbox(alignment = Pos.CENTER_LEFT, spacing = 5.0) {
-                h1(viewModel.nameProperty.value)
-                deleteButton()
-            }
-            subTitle(viewModel.subtitleProperty)
+            title = viewModel.nameProperty.value
+            subtitle = viewModel.subtitleProperty
+            buttons = listOf(deleteButton())
         }
         borderpane {
             padding = Insets(-5.0, 0.0, 10.0, 0.0)
@@ -114,7 +110,7 @@ class TopicView @Inject constructor(
         }.enableWhen(viewModel.consumerViewModel.isConsumingProperty.not())
 
     private fun EventTarget.deleteButton() =
-        confirmationButton("delete", "The topic \"${viewModel.nameProperty.value}\" will be removed.") {
+        confirmationButton("Delete", "The topic \"${viewModel.nameProperty.value}\" will be removed.") {
             viewModel.dispatch { delete() }
             closeTab()
         }.enableWhen(viewModel.consumerViewModel.isConsumingProperty.not())
diff --git a/lib/kafka/src/main/kotlin/insulator/kafka/SchemaRegistry.kt b/lib/kafka/src/main/kotlin/insulator/kafka/SchemaRegistry.kt
index df243790..ada86c10 100644
--- a/lib/kafka/src/main/kotlin/insulator/kafka/SchemaRegistry.kt
+++ b/lib/kafka/src/main/kotlin/insulator/kafka/SchemaRegistry.kt
@@ -2,16 +2,21 @@ package insulator.kafka
 
 import arrow.core.Either
 import arrow.core.computations.either
+import arrow.core.flatMap
 import insulator.helper.runCatchingE
 import insulator.kafka.model.Cluster
 import insulator.kafka.model.Schema
 import insulator.kafka.model.Subject
+import io.confluent.kafka.schemaregistry.avro.AvroSchema
 import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
 import io.confluent.kafka.schemaregistry.client.rest.RestService
 
 class SchemaRegistry(private val client: SchemaRegistryClient) {
 
+    fun register(subject: String, schema: String) = runCatchingE { AvroSchema(schema).also { it.validate() } }
+        .flatMap { client.runCatchingE { register(subject, it) } }.map { Unit }
+
     fun deleteSubject(subject: String) =
         client.runCatchingE { deleteSubject(subject) }.map { Unit }
 
diff --git a/lib/kafka/src/test/kotlin/insulator/kafka/SchemaRegistryTest.kt b/lib/kafka/src/test/kotlin/insulator/kafka/SchemaRegistryTest.kt
index 85530a89..1325a9df 100644
--- a/lib/kafka/src/test/kotlin/insulator/kafka/SchemaRegistryTest.kt
+++ b/lib/kafka/src/test/kotlin/insulator/kafka/SchemaRegistryTest.kt
@@ -2,7 +2,9 @@ package insulator.kafka
 
 import insulator.kafka.model.Cluster
 import insulator.kafka.model.SchemaRegistryConfiguration
+import io.confluent.kafka.schemaregistry.ParsedSchema
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
+import io.kotest.assertions.arrow.either.shouldBeLeft
 import io.kotest.assertions.arrow.either.shouldBeRight
 import io.kotest.core.spec.style.StringSpec
 import io.kotest.matchers.shouldNotBe
@@ -70,4 +72,30 @@ class SchemaRegistryTest : StringSpec({
         // assert
         res shouldBeRight { }
     }
+
+    "register an invalid schema return left" {
+        // arrange
+        val mockSchema = mockk<SchemaRegistryClient>()
+        val sut = SchemaRegistry(mockSchema)
+        // act
+        val res = sut.register("test", "{}")
+        // assert
+        res shouldBeLeft {}
+    }
+
+    "register a valid schema return true if accepted by the schema registry" {
+        // arrange
+        val mockSchema = mockk<SchemaRegistryClient> {
+            every { register(any(), any<ParsedSchema>()) } returns 1
+        }
+        val sut = SchemaRegistry(mockSchema)
+        val testSchema =
+            """
+            { "type": "record", "namespace": "com.example", "name": "FullName", "fields": [{ "name": "first", "type": "string" }] } 
+            """.trimIndent()
+        // act
+        val res = sut.register("test", testSchema)
+        // assert
+        res shouldBeRight {}
+    }
 })