Implement support for Protocol Buffer references (imports) -- continued

bin/avdlToAVSC.sh

@@ -1,3 +1,5 @@
+#!/bin/sh
+
 avdl_path=$1
 avsc_name=$2
 

docker-compose.yml

@@ -11,22 +11,22 @@ services:
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
     depends_on:
       - zooKeeper
-    image: 'confluentinc/cp-kafka:5.5.3'
+    image: 'confluentinc/cp-kafka:6.2.4'
     ports:
       - '9092:9092'
 
   zooKeeper: # https://hub.docker.com/r/confluentinc/cp-zookeeper
     environment:
       ZOOKEEPER_CLIENT_PORT: '2181'
-    image: 'confluentinc/cp-zookeeper:5.2.2'
+    image: 'confluentinc/cp-zookeeper:6.2.4'
     ports:
       - '2181:2181'
 
   schemaRegistry: # https://hub.docker.com/r/confluentinc/cp-schema-registry
     environment:
       SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zooKeeper:2181'
       SCHEMA_REGISTRY_HOST_NAME: localhost
-    image: 'confluentinc/cp-schema-registry:5.5.3'
+    image: 'confluentinc/cp-schema-registry:6.2.4'
     ports:
       - '8982:8081'
 

src/@types.ts

@@ -12,6 +12,7 @@ export enum SchemaType {
 export interface SchemaHelper {
   validate(schema: Schema): void
   getSubject(confluentSchema: ConfluentSchema, schema: Schema, separator: string): ConfluentSubject
+  referencedSchemas(schema: string): Promise<string[]>
 }
 
 export type AvroOptions = Partial<ForSchemaOptions>

src/AvroHelper.ts

@@ -49,6 +49,13 @@ export default class AvroHelper implements SchemaHelper {
     return subject
   }
 
+  public async referencedSchemas(
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    _schema: string,
+  ): Promise<string[]> {
+    return []
+  }
+
   private isRawAvroSchema(schema: ConfluentSchema | RawAvroSchema): schema is RawAvroSchema {
     const asRawAvroSchema = schema as RawAvroSchema
     return asRawAvroSchema.name != null && asRawAvroSchema.type != null

src/JsonHelper.ts

@@ -1,3 +1,4 @@
+/* eslint-disable @typescript-eslint/no-unused-vars */
 // @ts-nocheck
 import { Schema, SchemaHelper, ConfluentSubject, ConfluentSchema } from './@types'
 import { ConfluentSchemaRegistryError } from './errors'
@@ -14,4 +15,8 @@ export default class JsonHelper implements SchemaHelper {
   ): ConfluentSubject {
     throw new ConfluentSchemaRegistryError('not implemented yet')
   }
+
+  public async referencedSchemas(schema: string): Promise<string[]> {
+    return []
+  }
 }

src/ProtoHelper.ts

@@ -1,17 +1,39 @@
-// @ts-nocheck
 import { Schema, SchemaHelper, ConfluentSubject, ConfluentSchema } from './@types'
-import { ConfluentSchemaRegistryError } from './errors'
+import { ConfluentSchemaRegistryArgumentError, ConfluentSchemaRegistryError } from './errors'
+import { IParserResult, parse } from 'protobufjs'
 
 export default class ProtoHelper implements SchemaHelper {
-  public validate(schema: Schema): void {
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  public validate(_schema: Schema): void {
     return
   }
 
   public getSubject(
-    confluentSchema: ConfluentSchema,
-    schema: Schema,
-    separator: string,
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    _confluentSchema: ConfluentSchema,
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    _schema: Schema,
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    _separator: string,
   ): ConfluentSubject {
     throw new ConfluentSchemaRegistryError('not implemented yet')
   }
+
+  /**
+   * Get the schemas referenced by the provided schema.
+   *
+   * @param schema The schema to find references for
+   *
+   * @returns A list of imported/referenced schemas that is used by the given schema.
+   */
+  public async referencedSchemas(schema: string): Promise<string[]> {
+    let parsed: IParserResult
+    try {
+      parsed = parse(schema)
+    } catch (err) {
+      throw new ConfluentSchemaRegistryArgumentError(err)
+    }
+    const out: string[] = []
+    return out.concat(parsed.imports || []).concat(parsed.weakImports || [])
+  }
 }

src/ProtoSchema.ts

@@ -1,18 +1,62 @@
 import { Schema, ProtoOptions, ProtoConfluentSchema } from './@types'
-import protobuf from 'protobufjs'
+import protobuf, { AnyNestedObject } from 'protobufjs'
 import { IParserResult, ReflectionObject, Namespace, Type } from 'protobufjs/light'
 import {
   ConfluentSchemaRegistryArgumentError,
   ConfluentSchemaRegistryValidationError,
 } from './errors'
 
+interface NestedSchema {
+  [k: string]: AnyNestedObject
+}
+
 export default class ProtoSchema implements Schema {
   private message: Type
+  private root: protobuf.Root
 
-  constructor(schema: ProtoConfluentSchema, opts?: ProtoOptions) {
+  constructor(schema: ProtoConfluentSchema, opts?: ProtoOptions, references?: Schema[]) {
     const parsedMessage = protobuf.parse(schema.schema)
-    const root = parsedMessage.root
-    this.message = root.lookupType(this.getTypeName(parsedMessage, opts))
+    this.root = parsedMessage.root
+
+    if (references) {
+      const schemas = references as ProtoSchema[]
+      schemas.forEach(reference => {
+        // root.add() takes ownership over its input argument.
+        // add() can modify and extend the input schema, so we have to clone the reference to avoid polluting it.
+        const root = reference.root.toJSON().nested
+        if (!root) {
+          return
+        }
+
+        // make sure we add the items one-by-one to avoid adding duplicate items (which are not tolerated)
+        this.spreadReference(root)
+          .filter(item => {
+            const referencePath = this.referencePath(item)
+            return !this.root.lookup(referencePath)
+          })
+          .forEach(item => this.root.addJSON(item))
+      })
+    }
+
+    this.message = this.root.lookupType(this.getTypeName(parsedMessage, opts))
+  }
+
+  private referencePath(namespace: NestedSchema): string[] {
+    const [key, entry] = Object.entries(namespace)[0]
+    if ('nested' in entry && entry.nested) {
+      return [key, ...this.referencePath(entry.nested)]
+    }
+    return [key]
+  }
+
+  private spreadReference(namespace: NestedSchema): NestedSchema[] {
+    return Object.entries(namespace).flatMap(([key, entry]): NestedSchema[] => {
+      if ('nested' in entry && entry.nested) {
+        const nested: NestedSchema[] = this.spreadReference(entry.nested)
+        return nested.map((it: NestedSchema) => ({ [key]: { nested: it } }))
+      }
+      return [{ [key]: entry }]
+    })
   }
 
   private getNestedTypeName(parent: { [k: string]: ReflectionObject } | undefined): string {