Add SchemaRegistryClient, RestService, and testing (#1)

* Add SchemaRegistryClient, RestService, and testing

* Add new Makefile for schema registry

* Merging

* Revert to throwing exceptions

Author: Claimundefine

.eslintrc.js

@@ -0,0 +1,37 @@
+module.exports = {
+    "env": {
+        "browser": true,
+        "commonjs": true,
+        "es2021": true
+    },
+    "extends": "eslint:recommended",
+    "overrides": [
+        {
+            "env": {
+                "node": true
+            },
+            "files": [
+                ".eslintrc.{js,cjs}"
+            ],
+            "parserOptions": {
+                "sourceType": "script"
+            }
+        },
+        {
+            "files": ["*.ts"],
+            "parser": "@typescript-eslint/parser",
+            "parserOptions": {
+              "ecmaVersion": 2020,
+              "sourceType": "module"
+            },
+            "extends": [
+              "plugin:@typescript-eslint/recommended",
+            ]
+          }
+    ],
+    "parserOptions": {
+        "ecmaVersion": "latest"
+    },
+    "rules": {
+    }
+}

Makefile

@@ -88,4 +88,4 @@ release-patch:
 
 clean: node_modules/.dirstamp
 	@rm -f deps/librdkafka/config.h
-	@$(NODE-GYP) clean
+	@$(NODE-GYP) clean

Makefile.schemaregistry

@@ -0,0 +1,26 @@
+# Makefile.schemaregistry
+
+# Variables
+NODE ?= node
+ESLINT ?= ./node_modules/.bin/eslint
+JEST ?= ./node_modules/.bin/jest
+TS_NODE ?= ./node_modules/.bin/ts-node
+
+# Paths
+SRC_DIR = schemaregistry
+TEST_DIR = test/schemaregistry
+INTEG_DIR = e2e/schemaregistry
+
+# Tasks
+.PHONY: all lint test integtest
+
+all: lint test
+
+lint:
+	$(ESLINT) $(SRC_DIR) $(TEST_DIR) $(INTEG_DIR)
+
+test:
+	$(JEST) $(TEST_DIR)
+
+integtest:
+	$(JEST) $(INTEG_DIR)

e2e/schemaregistry/schemaregistry-client.spec.ts

@@ -0,0 +1,224 @@
+import { RestService } from '../../schemaregistry/rest-service';
+import {
+  Compatibility,
+  SchemaRegistryClient,
+  ServerConfig,
+  SchemaInfo,
+  SchemaMetadata,
+  Metadata
+} from '../../schemaregistry/schemaregistry-client';
+import { beforeEach, describe, expect, it } from '@jest/globals';
+
+/* eslint-disable @typescript-eslint/no-non-null-asserted-optional-chain */
+
+const baseUrls = ['http://localhost:8081'];
+const headers = { 'Content-Type': 'application/vnd.schemaregistry.v1+json' };
+const restService = new RestService(baseUrls, false);
+restService.setHeaders(headers);
+
+const basicAuth = Buffer.from('RBACAllowedUser-lsrc1:nohash').toString('base64');
+restService.setAuth(basicAuth);
+
+restService.setTimeout(10000);
+
+let schemaRegistryClient: SchemaRegistryClient;
+const testSubject = 'integ-test-subject';
+const testServerConfigSubject = 'integ-test-server-config-subject';
+
+const schemaString: string = JSON.stringify({
+  type: 'record',
+  name: 'User',
+  fields: [
+    { name: 'name', type: 'string' },
+    { name: 'age', type: 'int' },
+  ],
+});
+
+const metadata: Metadata = {
+  properties: {
+    owner: 'Bob Jones',
+    email: '[email protected]',
+  },
+};
+
+const schemaInfo: SchemaInfo = {
+  schema: schemaString,
+  metadata: metadata,
+};
+
+const backwardCompatibleSchemaString: string = JSON.stringify({
+  type: 'record',
+  name: 'User',
+  fields: [
+    { name: 'name', type: 'string' },
+    { name: 'age', type: 'int' },
+    { name: 'email', type: 'string', default: "" },
+  ],
+});
+
+const backwardCompatibleMetadata: Metadata = {
+  properties: {
+    owner: 'Bob Jones2',
+    email: '[email protected]',
+  },
+};
+
+const backwardCompatibleSchemaInfo: SchemaInfo = {
+  schema: backwardCompatibleSchemaString,
+  schemaType: 'AVRO',
+  metadata: backwardCompatibleMetadata,
+};
+
+describe('SchemaRegistryClient Integration Test', () => {
+
+  beforeEach(async () => {
+    schemaRegistryClient = new SchemaRegistryClient(restService);
+    const subjects: string[] = await schemaRegistryClient.getAllSubjects();
+
+    if (subjects && subjects.includes(testSubject)) {
+      await schemaRegistryClient.deleteSubject(testSubject);
+      await schemaRegistryClient.deleteSubject(testSubject, true);
+    }
+
+    if (subjects && subjects.includes(testServerConfigSubject)) {
+      await schemaRegistryClient.deleteSubject(testServerConfigSubject);
+      await schemaRegistryClient.deleteSubject(testServerConfigSubject, true);
+    }
+  });
+
+  it('should register, retrieve, and delete a schema', async () => {
+    // Register a schema
+    const registerResponse:  SchemaMetadata = await schemaRegistryClient.registerFullResponse(testSubject, schemaInfo);
+    expect(registerResponse).toBeDefined();
+
+    const schemaId = registerResponse?.id!;
+    const version = registerResponse?.version!;
+
+    const getSchemaResponse: SchemaInfo = await schemaRegistryClient.getBySubjectAndId(testSubject, schemaId);
+    expect(getSchemaResponse).toEqual(schemaInfo);
+
+    const getIdResponse: number = await schemaRegistryClient.getId(testSubject, schemaInfo);
+    expect(getIdResponse).toEqual(schemaId);
+
+    // Delete the schema
+    const deleteSubjectResponse: number = await schemaRegistryClient.deleteSubjectVersion(testSubject, version);
+    expect(deleteSubjectResponse).toEqual(version);
+
+    const permanentDeleteSubjectResponse: number = await schemaRegistryClient.deleteSubjectVersion(testSubject, version, true);
+    expect(permanentDeleteSubjectResponse).toEqual(version);
+  });
+
+  it('Should get all versions and a specific version of a schema', async () => {
+    // Register a schema
+    const registerResponse:  SchemaMetadata = await schemaRegistryClient.registerFullResponse(testSubject, schemaInfo);
+    expect(registerResponse).toBeDefined();
+
+    const version = registerResponse?.version!;
+
+    const getVersionResponse: number = await schemaRegistryClient.getVersion(testSubject, schemaInfo);
+    expect(getVersionResponse).toEqual(version);
+
+    const allVersionsResponse: number[] = await schemaRegistryClient.getAllVersions(testSubject);
+    expect(allVersionsResponse).toEqual([version]);
+  });
+
+  it('Should get schema metadata', async () => {
+    // Register a schema
+    const registerResponse:  SchemaMetadata = await schemaRegistryClient.registerFullResponse(testSubject, schemaInfo);
+    expect(registerResponse).toBeDefined();
+
+    const schemaVersion: number = registerResponse?.version!;
+
+    const registerResponse2:  SchemaMetadata = await schemaRegistryClient.registerFullResponse(testSubject, backwardCompatibleSchemaInfo);
+    expect(registerResponse2).toBeDefined();
+
+    const schemaMetadata: SchemaMetadata = {
+      id: registerResponse?.id!,
+      version: schemaVersion,
+      schema: schemaInfo.schema,
+      subject: testSubject,
+      metadata: metadata,
+    };
+
+    const schemaMetadata2: SchemaMetadata = {
+      id: registerResponse2?.id!,
+      version: registerResponse2?.version!,
+      schema: backwardCompatibleSchemaInfo.schema,
+      subject: testSubject,
+      metadata: backwardCompatibleMetadata,
+    };
+
+    const getLatestMetadataResponse:  SchemaMetadata = await schemaRegistryClient.getLatestSchemaMetadata(testSubject);
+    expect(schemaMetadata2).toEqual(getLatestMetadataResponse);
+
+    const getMetadataResponse:  SchemaMetadata = await schemaRegistryClient.getSchemaMetadata(testSubject, schemaVersion);
+    expect(schemaMetadata).toEqual(getMetadataResponse);
+  });
+
+  it('Should test compatibility for a version and subject, getting and updating', async () => {
+    const registerResponse:  SchemaMetadata = await schemaRegistryClient.registerFullResponse(testSubject, schemaInfo);
+    expect(registerResponse).toBeDefined();
+
+    const version = registerResponse?.version!;
+
+    const updateCompatibilityResponse: Compatibility = await schemaRegistryClient.updateCompatibility(testSubject, Compatibility.BackwardTransitive);
+    expect(updateCompatibilityResponse).toEqual(Compatibility.BackwardTransitive);
+
+    const getCompatibilityResponse: Compatibility = await schemaRegistryClient.getCompatibility(testSubject);
+    expect(getCompatibilityResponse).toEqual(Compatibility.BackwardTransitive);
+
+    const testSubjectCompatibilityResponse: boolean = await schemaRegistryClient.testSubjectCompatibility(testSubject, backwardCompatibleSchemaInfo);
+    expect(testSubjectCompatibilityResponse).toEqual(true);
+
+    const testCompatibilityResponse: boolean = await schemaRegistryClient.testCompatibility(testSubject, version, backwardCompatibleSchemaInfo);
+    expect(testCompatibilityResponse).toEqual(true);
+  });
+
+  it('Should update and get default compatibility', async () => {
+    const updateDefaultCompatibilityResponse: Compatibility = await schemaRegistryClient.updateDefaultCompatibility(Compatibility.Full);
+    expect(updateDefaultCompatibilityResponse).toEqual(Compatibility.Full);
+
+    const getDefaultCompatibilityResponse: Compatibility = await schemaRegistryClient.getDefaultCompatibility();
+    expect(getDefaultCompatibilityResponse).toEqual(Compatibility.Full);
+  });
+
+  it('Should update and get subject Config', async () => {
+    const subjectConfigRequest: ServerConfig = {
+      compatibility: Compatibility.Full,
+      normalize: true
+    };
+
+    const subjectConfigResponse: ServerConfig = {
+      compatibilityLevel: Compatibility.Full,
+      normalize: true
+    };
+
+    const registerResponse:  SchemaMetadata = await schemaRegistryClient.registerFullResponse(testServerConfigSubject, schemaInfo);
+    expect(registerResponse).toBeDefined();
+
+    const updateConfigResponse: ServerConfig = await schemaRegistryClient.updateConfig(testServerConfigSubject, subjectConfigRequest);
+    expect(updateConfigResponse).toBeDefined();
+
+    const getConfigResponse: ServerConfig = await schemaRegistryClient.getConfig(testServerConfigSubject);
+    expect(getConfigResponse).toEqual(subjectConfigResponse);
+  });
+
+  it('Should get and set default Config', async () => {
+    const serverConfigRequest: ServerConfig = {
+      compatibility: Compatibility.Full,
+      normalize: false
+    };
+
+    const serverConfigResponse: ServerConfig = {
+      compatibilityLevel: Compatibility.Full,
+      normalize: false
+    };
+
+    const updateDefaultConfigResponse: ServerConfig = await schemaRegistryClient.updateDefaultConfig(serverConfigRequest);
+    expect(updateDefaultConfigResponse).toBeDefined();
+
+    const getDefaultConfigResponse: ServerConfig = await schemaRegistryClient.getDefaultConfig();
+    expect(getDefaultConfigResponse).toEqual(serverConfigResponse);
+  });
+
+});

jest.config.js

@@ -0,0 +1,8 @@
+module.exports = {
+    preset: 'ts-jest',
+    testEnvironment: 'node',
+    testMatch: ['**/test/**/*.ts', '**/e2e/**/*.ts'],
+    transform: {
+      '^.+\\.tsx?$': 'ts-jest',
+    },
+  };

package-lock.json

@@ -12,7 +12,8 @@
     "test": "make test",
     "install": "node-pre-gyp install --fallback-to-build",
     "prepack": "node ./ci/prepublish.js",
-    "test:types": "tsc -p ."
+    "test:types": "tsc -p .",
+    "test:schemaregistry": "make -f Makefile.schemaregistry test"
   },
   "binary": {
     "module_name": "confluent-kafka-javascript",

package.json

@@ -0,0 +1,72 @@
+import axios, { AxiosInstance, AxiosRequestConfig, AxiosResponse } from 'axios';
+
+/*
+ * Confluent-Schema-Registry-TypeScript - Node.js wrapper for Confluent Schema Registry
+ *
+ * Copyright (c) 2024 Confluent, Inc.
+ *
+ * This software may be modified and distributed under the terms
+ * of the MIT license.  See the LICENSE.txt file for details.
+ */
+
+export class RestService {
+  private client: AxiosInstance
+
+  constructor(baseUrls: string[], isForward = false) {
+    this.client = axios.create({
+      baseURL: baseUrls[0], // Use the first base URL as the default
+      timeout: 5000, // Default timeout
+      headers: { 'Content-Type': 'application/vnd.schemaregistry.v1+json' },
+    })
+
+    if (isForward) {
+      this.client.defaults.headers.common['X-Forward'] = 'true'
+    }
+  }
+
+  public async sendHttpRequest<T>(
+    url: string,
+    method: 'GET' | 'POST' | 'PUT' | 'DELETE',
+    data?: any, // eslint-disable-line @typescript-eslint/no-explicit-any
+    config?: AxiosRequestConfig,
+  ): Promise<AxiosResponse<T>> {
+    try {
+      const response = await this.client.request<T>({
+        url,
+        method,
+        data,
+        ...config,
+      })
+      return response
+    } catch (error) {
+      if (axios.isAxiosError(error) && error.response) {
+        throw new Error(`HTTP error: ${error.response.status} - ${error.response.data}`)
+      } else {
+        const err = error as Error;
+        throw new Error(`Unknown error: ${err.message}`)
+      }
+    }
+  }
+
+  public setHeaders(headers: Record<string, string>): void {
+    this.client.defaults.headers.common = { ...this.client.defaults.headers.common, ...headers }
+  }
+
+  public setAuth(basicAuth?: string, bearerToken?: string): void {
+    if (basicAuth) {
+      this.client.defaults.headers.common['Authorization'] = `Basic ${basicAuth}`
+    }
+
+    if (bearerToken) {
+      this.client.defaults.headers.common['Authorization'] = `Bearer ${bearerToken}`
+    }
+  }
+
+  public setTimeout(timeout: number): void {
+    this.client.defaults.timeout = timeout
+  }
+
+  public setBaseURL(baseUrl: string): void {
+    this.client.defaults.baseURL = baseUrl
+  }
+}