Added basic E2E test

.github/workflows/test.yml

@@ -18,6 +18,9 @@ jobs:
       with:
         node-version: ${{ matrix.node-version }}
     - run: npm install
+    - name: Start Docker containers for Zookeeper, Kafka and Schema Registry
+      run: npm run kafka:start
+    - run: sleep 60
     - run: npm test
       env:
         CI: true

.npmignore

@@ -2,3 +2,5 @@ src/
 tsconfig.json
 .github/
 .gitignore
+kafka/
+babel.config.js

kafka/README.md

@@ -0,0 +1,3 @@
+# KAFKA
+
+A kafka environment to run E2E tests on. Based on [cp-all-in-one](https://github.com/confluentinc/cp-all-in-one/blob/5.5.1-post/cp-all-in-one-community/docker-compose.yml)

kafka/docker-compose.yml

@@ -0,0 +1,40 @@
+---
+version: '2'
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:5.5.1
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  broker:
+    image: confluentinc/cp-kafka:5.5.1
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+      - "9092:9092"
+      - "9101:9101"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_JMX_PORT: 9101
+
+  schema-registry:
+    image: confluentinc/cp-schema-registry:5.5.1
+    depends_on:
+      - zookeeper
+      - broker
+    ports:
+      - "8081:8081"
+    environment:
+      SCHEMA_REGISTRY_HOST_NAME: schema-registry
+      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'

package.json

@@ -5,14 +5,15 @@
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
   "scripts": {
-    "prepare": "tsc",
+    "prepare": "tsc --project tsconfig.build.json",
     "test": "jest",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
-    "kafka:start": "docker-compose up -d -f ./kafka/docker-compose.yml",
+    "kafka:start": "docker-compose -f ./kafka/docker-compose.yml up -d",
+    "kafka:stop": "docker-compose -f ./kafka/docker-compose.yml down",
     "test:watch": "jest --watch",
     "test:cov": "jest --coverage",
     "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
-    "test:e2e": "jest --config ./jest-e2e.json"
+    "test:e2e": "jest --config ./test/jest-e2e.json --detectOpenHandles --runInBand"
   },
   "repository": {
     "type": "git",
@@ -55,5 +56,19 @@
     "prettier": "^2.1.2",
     "ts-jest": "^26.3.0",
     "typescript": "^4.0.2"
+  },
+  "jest": {
+    "moduleFileExtensions": [
+      "js",
+      "json",
+      "ts"
+    ],
+    "rootDir": "src",
+    "testRegex": ".spec.ts$",
+    "transform": {
+      "^.+\\.(t|j)s$": "ts-jest"
+    },
+    "coverageDirectory": "../coverage",
+    "testEnvironment": "node"
   }
 }

test/app.e2e-spec.ts

@@ -0,0 +1,56 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { INestMicroservice } from '@nestjs/common';
+import AppModule from './e2e/app/config.app';
+import { TestConsumer, TOPIC_NAME } from './e2e/app/test.controller';
+
+describe('AppModule (e2e)', () => {
+  let app: INestMicroservice;
+
+  beforeAll(async () => {
+    const moduleFixture: TestingModule = await Test.createTestingModule({
+      imports: [AppModule],
+    }).compile();
+
+    app = moduleFixture.createNestMicroservice({});
+    app.enableShutdownHooks();
+    await app.listenAsync();
+  });
+
+  afterAll(async() => {
+    await app.close();
+  });
+
+  it("should give kafka some time", done => {
+    setTimeout(done, 2500);
+  });
+
+  it('can produce JSON messages', async () => {
+    const cont = await app.resolve(TestConsumer);
+
+    return cont.sendMessage([{
+      value: 'Hello World!',
+      timestamp: Date.now().toString(),
+    }]).then((value) => {
+      console.log(value);
+      expect(value).toBe([{
+        topicName: TOPIC_NAME,
+        partition: 0,
+        errorCode: 0,
+        baseOffset: '0',
+        logAppendTime: '-1',
+        logStartOffset: '0'
+      }]);
+    });
+  });
+
+  // it('can accept JSON message', () => {
+
+
+  // });
+
+  // it('can produce AVRO message', () => {
+  // });
+
+  // it('can accept AVRO message', () => {
+  // });
+});

test/e2e/app/config.app.ts

@@ -0,0 +1,40 @@
+import { Module } from '@nestjs/common';
+import { KafkaModule, KafkaAvroResponseDeserializer, KafkaAvroMessageSerializer } from "../../../src";
+import { TestConsumer } from "./test.controller";
+
+@Module({
+  imports: [
+    KafkaModule.register([
+      {
+        name: 'KAFKA_SERVICE',
+        options: {
+          client: {
+            clientId: 'test-e2e',
+            brokers: ['localhost:9092'],
+            retry: {
+              retries: 0,
+              initialRetryTime: 1,
+            },
+          },
+          consumer: {
+            groupId: 'test-e2e-consumer',
+          },
+          deserializer: new KafkaAvroResponseDeserializer({
+            host: 'http://localhost:8081/'
+          }),
+          // serializer: new KafkaAvroMessageSerializer({
+          //   config: {
+          //     host: 'http://localhost:8081/'
+          //   },
+          //   schemas: [
+          //     './schema.avsc'
+          //   ]
+          // }),
+          consumeFromBeginning: true
+        }
+      },
+    ]),
+    TestConsumer
+  ],
+})
+export default class AppModule {}

test/e2e/app/schema.avsc

@@ -0,0 +1,39 @@
+{
+  "name": "Value",
+  "namespace": "test.topic",
+  "type": "record",
+  "fields": [
+    {
+      "name": "id",
+      "type": "int"
+    },
+    {
+      "name": "metadataId",
+      "type": "int"
+    },
+    {
+      "name": "objectId",
+      "type": "int"
+    },
+    {
+      "name": "value",
+      "type": "string"
+    },
+    {
+      "default": null,
+      "name": "__table",
+      "type": [
+        "null",
+        "string"
+      ]
+    },
+    {
+      "default": null,
+      "name": "__deleted",
+      "type": [
+        "null",
+        "string"
+      ]
+    }
+  ],
+}

test/e2e/app/test.controller.ts

@@ -0,0 +1,33 @@
+import { Inject, Post } from '@nestjs/common';
+import { Payload } from "@nestjs/microservices";
+import { ProducerRecord } from 'kafkajs';
+import { SubscribeTo, KafkaService } from "../../../dist";
+
+export const TOPIC_NAME = 'test.topic';
+
+export class TestConsumer {
+
+  constructor(
+    @Inject('KAFKA_SERVICE') private client: KafkaService
+  ) {
+  }
+
+  onModuleInit(): void {
+    this.client.subscribeToResponseOf(TOPIC_NAME, this)
+  }
+
+  @SubscribeTo(TOPIC_NAME)
+  async message(@Payload() data: any): Promise<void> {
+    console.log(data);
+  }
+
+  @Post()
+  async sendMessage(messages: ProducerRecord["messages"]) {
+    console.log('sending!!', messages);
+    return await this.client.send({
+      topic: TOPIC_NAME,
+
+      messages,
+    });
+  }
+}

test/jest-e2e.json

@@ -0,0 +1,9 @@
+{
+  "moduleFileExtensions": ["js", "json", "ts"],
+  "rootDir": ".",
+  "testEnvironment": "node",
+  "testRegex": ".e2e-spec.ts$",
+  "transform": {
+    "^.+\\.(t|j)s$": "ts-jest"
+  }
+}

tsconfig.build.json

@@ -0,0 +1,4 @@
+{
+  "extends": "./tsconfig.json",
+  "exclude": ["node_modules", "test", "dist", "**/*spec.ts", "examples"]
+}

tsconfig.json

@@ -2,18 +2,14 @@
   "compilerOptions": {
     "module": "commonjs",
     "declaration": true,
-    "noImplicitAny": false,
-    "noUnusedLocals": false,
     "removeComments": true,
-    "noLib": false,
     "emitDecoratorMetadata": true,
     "experimentalDecorators": true,
-    "target": "es6",
+    "allowSyntheticDefaultImports": true,
+    "target": "es2017",
     "sourceMap": true,
-    "outDir": "dist",
-    "lib": ["es7"],
-    "skipLibCheck": true
-  },
-  "include": ["src/**/*"],
-  "exclude": ["node_modules"]
+    "outDir": "./dist",
+    "baseUrl": "./",
+    "incremental": true
+  }
 }