Merge pull request #113 from MikeDalby/issue-112-fix-nested-protobuf-…

…types breaking: Issue-112 Fix Default Nested Protobuf Types
kafkajs · May 10, 2021 · f9bd3ed · f9bd3ed
2 parents 10d38ff + 773bce2
commit f9bd3ed
Show file tree

Hide file tree

Showing 3 changed files with 72 additions and 1 deletion.
diff --git a/fixtures/proto/encodedNestedV2.ts b/fixtures/proto/encodedNestedV2.ts
@@ -0,0 +1 @@
+export default [0, 0, 0, 0, 32, 10, 10, 100, 97, 116, 97, 45, 118, 97, 108, 117, 101, 18, 17, 10, 15, 115, 111, 109, 101, 70, 105, 101, 108, 100, 45, 118, 97, 108, 117, 101, 26, 22, 10, 20, 115, 111, 109, 101, 79, 116, 104, 101, 114, 70, 105, 101, 108, 100, 45, 118, 97, 108, 117, 101 ]
diff --git a/src/ProtoSchema.ts b/src/ProtoSchema.ts
@@ -19,7 +19,9 @@ export default class ProtoSchema implements Schema {
     if (!parent) throw new ConfluentSchemaRegistryArgumentError('no nested fields')
     const keys = Object.keys(parent)
     const reflection = parent[keys[0]]
-    if (reflection instanceof Namespace && reflection.nested)
+
+    // Traverse down the nested Namespaces until we find a message Type instance (which extends Namespace)
+    if (reflection instanceof Namespace && !(reflection instanceof Type) && reflection.nested)
       return this.getNestedTypeName(reflection.nested)
     return keys[0]
   }

diff --git a/src/SchemaRegistry.newApi.spec.ts b/src/SchemaRegistry.newApi.spec.ts
@@ -13,6 +13,7 @@ import { COMPATIBILITY, DEFAULT_API_CLIENT_ID } from './constants'
 import encodedAnotherPersonV2Avro from '../fixtures/avro/encodedAnotherPersonV2'
 import encodedAnotherPersonV2Json from '../fixtures/json/encodedAnotherPersonV2'
 import encodedAnotherPersonV2Proto from '../fixtures/proto/encodedAnotherPersonV2'
+import encodedNestedV2Proto from '../fixtures/proto/encodedNestedV2'
 import wrongMagicByte from '../fixtures/wrongMagicByte'
 
 const REGISTRY_HOST = 'http://localhost:8982'
@@ -492,5 +493,72 @@ describe('SchemaRegistry - new Api', () => {
 
       expect(data).toEqual(payload)
     })
+
+    describe('nested message types tests', () => {
+      const v4 = `
+      syntax = "proto2";
+      package com.org.domain.fixtures;
+      message OuterMessageType {
+        required string data = 1;
+        required InnerMessageType1 innerMessageType1 = 2;
+        required InnerMessageType2 innerMessageType2 = 3;
+
+        message InnerMessageType1 {
+          required string someField = 1;
+        }
+        message InnerMessageType2 {
+          required string someOtherField = 1;
+        }
+      }
+      `,
+        type = SchemaType.PROTOBUF,
+        nestedPayload = {
+          data: 'data-value',
+          innerMessageType1: {
+            someField: 'someField-value',
+          },
+          innerMessageType2: {
+            someOtherField: 'someOtherField-value',
+          },
+        }
+
+      beforeAll(() => {
+        schemaRegistry = new SchemaRegistry(schemaRegistryArgs)
+      })
+
+      it('encodes', async () => {
+        const confluentSchemaV4: ConfluentSchema = {
+          type,
+          schema: v4,
+        }
+
+        const schema4 = await schemaRegistry.register(confluentSchemaV4, {
+          subject: `${type}_test4`,
+        })
+
+        const data = await schemaRegistry.encode(schema4.id, nestedPayload)
+
+        expect(data).toMatchConfluentEncodedPayload({
+          registryId: schema4.id,
+          payload: Buffer.from(encodedNestedV2Proto),
+        })
+      })
+
+      it('decodes', async () => {
+        const confluentSchemaV4: ConfluentSchema = {
+          type,
+          schema: v4,
+        }
+
+        const schema4 = await schemaRegistry.register(confluentSchemaV4, {
+          subject: `${type}_test4`,
+        })
+
+        const buffer = Buffer.from(await schemaRegistry.encode(schema4.id, nestedPayload))
+        const data = await schemaRegistry.decode(buffer)
+
+        expect(data).toEqual(nestedPayload)
+      })
+    })
   })
 })