core: add check-schema

core/pool.rkt

@@ -36,6 +36,7 @@
  pool-activate-registry
  pool-deactivate-registry
  pool-get-schema
+ pool-check-schema
  pool-create-schema
  pool-delete-schema
  pool-open-iterator
@@ -334,6 +335,14 @@
                          (sr:get-schema r name))))
                     (state-add-req s (req result res-ch nack))]
 
+                   [`(check-schema ,res-ch ,nack ,id ,name ,schema)
+                    (define result
+                      (call-with-state-schema-registry*
+                       s id
+                       (lambda (r)
+                         (sr:check-schema r name schema))))
+                    (state-add-req s (req result res-ch nack))]
+
                    [`(create-schema ,res-ch ,nack ,id ,name ,type ,schema)
                     (define result
                       (call-with-state-schema-registry*
@@ -474,6 +483,9 @@
 (define (pool-get-schema id name [p (current-pool)])
   (force (sync (pool-send p get-schema id name))))
 
+(define (pool-check-schema id name schema [p (current-pool)])
+  (force (sync (pool-send p check-schema id name schema))))
+
 (define (pool-create-schema id name type schema [p (current-pool)])
   (force (sync (pool-send p create-schema id name type schema))))
 

core/schema-registry.rkt

@@ -98,6 +98,11 @@
                         [in-workspace id : UVarint] : Schema)
   (pool-get-schema id name))
 
+(define-rpc (check-schema [named name : String]
+                          [with-updated-schema schema : String]
+                          [in-workspace id : UVarint])
+  (pool-check-schema id name schema))
+
 (define-rpc (create-schema [named name : String]
                            [of-type type : Symbol]
                            [with-schema schema : String]

core/schema-registry/confluent.rkt

@@ -37,6 +37,17 @@
       #:version (impl:Schema-version schema)
       #:schema (impl:Schema-schema schema)))
 
+   (define (check-schema self name schema)
+     (define client
+       (confluent-registry-client self))
+     (define current-schema
+       (impl:get-subject-version client name 'latest))
+     (define the-schema
+       (impl:make-Schema
+        #:type (impl:Schema-type current-schema)
+        #:schema schema))
+     (impl:check-schema client name the-schema))
+
    (define (create-schema self name type schema)
      (define client
        (confluent-registry-client self))

core/schema-registry/generic.rkt

@@ -7,13 +7,15 @@
  gen:registry
  get-schemas
  get-schema
+ check-schema
  create-schema
  delete-schema
  decode-record)
 
 (define-generics registry
   {get-schemas registry}
   {get-schema registry name}
+  {check-schema registry name schema}
   {create-schema registry name type schema}
   {delete-schema registry name}
   {decode-record registry record})