feat(BE): Get latest schema for a given subject

src-tauri/Cargo.toml

@@ -20,6 +20,7 @@ serde = { version = "1.0", features = ["derive"] }
 tauri = { version = "1.1", features = ["api-all"] }
 rdkafka = { version = "0.27", features = ["cmake-build", "ssl-vendored"] }
 reqwest = { version = "0.11", features = ["json"] }
+url = { version = "2", features = ["serde"] }
 tokio = { version = "1", features = ["full"] }
 
 [features]

src-tauri/src/error.rs

@@ -44,4 +44,13 @@ impl From<reqwest::Error> for TauriError {
             message: error.to_string(),
         }
     }
+}
+
+impl From<url::ParseError> for TauriError {
+    fn from(error: url::ParseError) -> Self {
+        TauriError {
+            error_type: "Url parsing".to_owned(),
+            message: error.to_string(),
+        }
+    }
 }

src-tauri/src/main.rs

@@ -8,7 +8,7 @@ mod error;
 use crate::{
     kafka::list_topics,
     configuration::{ get_configuration, write_configuration },
-    schema_registry::{ list_subjects },
+    schema_registry::{ list_subjects, get_schema },
 };
 
 fn main() {
@@ -19,7 +19,8 @@ fn main() {
                 get_configuration,
                 write_configuration,
                 list_topics,
-                list_subjects
+                list_subjects,
+                get_schema
             ]
         )
         .run(tauri::generate_context!())

src-tauri/src/schema_registry/mod.rs

@@ -1,19 +1,49 @@
-use serde::{ de::DeserializeOwned };
+use url::Url;
+use serde::{ de::DeserializeOwned, Deserialize, Serialize };
 
-use crate::{ configuration::model::SchemaRegistry, error::Result };
+use crate::{ configuration::model::SchemaRegistry, error::{ Result, TauriError } };
 
-async fn get<T: DeserializeOwned>(url: String, config: SchemaRegistry) -> Result<T> {
+async fn get<T: DeserializeOwned>(url: String, config: &SchemaRegistry) -> Result<T> {
+    println!("{}", url);
     let client = reqwest::Client::new();
     let mut request = client.get(url);
-    if let Some(username) = config.username {
-        request = request.basic_auth(username, config.password);
+    if let Some(username) = &config.username {
+        request = request.basic_auth(username, config.password.as_ref());
     }
-    let res = request.send().await?.json().await?;
+    let response = request.send().await?;
+    let res = response.json().await?;
     Ok(res)
 }
 
 #[tauri::command]
 pub async fn list_subjects(config: SchemaRegistry) -> Result<Vec<String>> {
-    let res = get(format!("{:}/subjects", config.endpoint), config).await?;
+    let url = Url::parse(&config.endpoint)?.join("subjects")?;
+    let res = get(url.to_string(), &config).await?;
     Ok(res)
+}
+
+#[tauri::command]
+pub async fn get_schema(subject_name: String, config: SchemaRegistry) -> Result<Schema> {
+    let url = Url::parse(&config.endpoint)?.join(
+        format!("/subjects/{}/versions/", subject_name).as_str()
+    )?;
+    let versions: Vec<i32> = get(url.to_string(), &config).await?;
+    if let Some(latest_version) = versions.iter().max() {
+        let latest_schema_url = url.join(&latest_version.to_string())?;
+        let latest_schema: Schema = get(latest_schema_url.to_string(), &config).await?;
+        Ok(latest_schema)
+    } else {
+        Err(TauriError {
+            error_type: "Schema registry".to_string(),
+            message: format!("No versions found for subject {:}", subject_name),
+        })
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct Schema {
+    subject: String,
+    id: i32,
+    version: i32,
+    schema: String,
 }

src/pages/schema-registry/main.tsx

@@ -1,19 +1,25 @@
-import { Group } from "@mantine/core";
+import { Group, Text } from "@mantine/core";
 import { useState } from "react";
+import { useAppState } from "../../providers";
 import { Schema } from "./schema";
 import { SchemaList } from "./schema-list";
 
 export const SchemasPage = () => {
+  const { appState } = useAppState();
   const [state, setState] = useState<{ activeSchema?: string }>({});
   const { activeSchema } = state;
-  return (
-    <Group grow={true} align={"stretch"} position={"center"} noWrap={true}>
-      <SchemaList
-        onTopicSelected={(activeSchema) => {
-          setState({ ...state, activeSchema: activeSchema });
-        }}
-      />
-      {activeSchema && <Schema schemaName={activeSchema} />}
-    </Group>
-  );
+  const schemaRegistry = appState.activeCluster?.schemaRegistry;
+  if (schemaRegistry && schemaRegistry.endpoint) {
+    return (
+      <Group grow={true} align={"stretch"} position={"center"} noWrap={true}>
+        <SchemaList
+          schemaRegistry={schemaRegistry}
+          onTopicSelected={(activeSchema) => {
+            setState({ ...state, activeSchema: activeSchema });
+          }}
+        />
+        {activeSchema && <Schema schemaRegistry={schemaRegistry} schemaName={activeSchema} />}
+      </Group>
+    );
+  } else return <Text>Missing schema registry configuration</Text>;
 };

src/pages/schema-registry/schema-list.tsx

@@ -1,43 +1,39 @@
 import { invoke } from "@tauri-apps/api";
 import { useMemo, useState } from "react";
-import { Cluster } from "../../models/kafka";
-import { useAppState, notifyAlert, notifySuccess } from "../../providers";
+import { SchemaRegistry } from "../../models/kafka";
+import { notifyAlert, notifySuccess } from "../../providers";
 import { format, TauriError } from "../../tauri";
 import { ItemList } from "../common";
 
-function getSchemaNamesList(cluster: Cluster): Promise<string[]> {
-  return invoke<string[]>("list_subjects", { config: cluster.schemaRegistry });
+function getSchemaNamesList(config: SchemaRegistry): Promise<string[]> {
+  return invoke<string[]>("list_subjects", { config });
 }
 
 export const SchemaList = ({
+  schemaRegistry,
   onTopicSelected,
 }: {
+  schemaRegistry: SchemaRegistry;
   onTopicSelected: (topicName: string) => void;
 }) => {
-  const { appState } = useAppState();
   const [state, setState] = useState<{ schemas: string[]; search?: string; loading: boolean }>({
     schemas: [],
     loading: true,
   });
 
   const updateSchemasList = () => {
-    if (appState.activeCluster) {
-      setState({ ...state, loading: true });
-      getSchemaNamesList(appState.activeCluster)
-        .then((schemas) => setState({ schemas, loading: false }))
-        .then((_) => notifySuccess("List of schemas successfully retrieved"))
-        .catch((err: TauriError) => {
-          notifyAlert(
-            `Unable to retrieve the list of schemas for "${appState.activeCluster?.name}"`,
-            format(err)
-          );
-          setState({ schemas: [], loading: false });
-        });
-    }
+    setState({ ...state, loading: true });
+    getSchemaNamesList(schemaRegistry)
+      .then((schemas) => setState({ schemas, loading: false }))
+      .then((_) => notifySuccess("List of schemas successfully retrieved"))
+      .catch((err: TauriError) => {
+        notifyAlert(`Unable to retrieve the list of schemas.`, format(err));
+        setState({ schemas: [], loading: false });
+      });
   };
 
   // eslint-disable-next-line react-hooks/exhaustive-deps
-  useMemo(() => updateSchemasList(), [appState.activeCluster]);
+  useMemo(() => updateSchemasList(), [schemaRegistry]);
 
   return (
     <ItemList

src/pages/schema-registry/schema.tsx

@@ -1,18 +1,36 @@
 import { ActionIcon, Container, Divider, Group, Title, Tooltip } from "@mantine/core";
 import { IconInfoCircle } from "@tabler/icons";
+import { invoke } from "@tauri-apps/api";
+import { useMemo } from "react";
+import { SchemaRegistry } from "../../models/kafka";
 
-export const Schema = ({ schemaName }: { schemaName: string }) => (
-  <Container style={{ width: "100%" }}>
-    <Group position={"apart"}>
-      <Title>{schemaName}</Title>
-      <Group>
-        <Tooltip label="Topic info">
-          <ActionIcon>
-            <IconInfoCircle />
-          </ActionIcon>
-        </Tooltip>
+const getLatestSchema = (subjectName: string, config: SchemaRegistry) =>
+  invoke("get_schema", { subjectName, config }).catch((err) => console.error(err));
+
+export const Schema = ({
+  schemaName,
+  schemaRegistry,
+}: {
+  schemaName: string;
+  schemaRegistry: SchemaRegistry;
+}) => {
+  useMemo(async () => {
+    const lastSchema = await getLatestSchema(schemaName, schemaRegistry);
+    console.log(lastSchema);
+  }, [schemaName]);
+  return (
+    <Container style={{ width: "100%" }}>
+      <Group position={"apart"}>
+        <Title>{schemaName}</Title>
+        <Group>
+          <Tooltip label="Topic info">
+            <ActionIcon>
+              <IconInfoCircle />
+            </ActionIcon>
+          </Tooltip>
+        </Group>
       </Group>
-    </Group>
-    <Divider mt={10} />
-  </Container>
-);
+      <Divider mt={10} />
+    </Container>
+  );
+};