Feature: Spring Kafka

README.md

@@ -66,15 +66,15 @@ For complete instructions on generating the Java and subsequently using it to se
 
 You can configure this template by passing different parameters in the Generator CLI: `-p PARAM1_NAME=PARAM1_VALUE -p PARAM2_NAME=PARAM2_VALUE`
 
-Name | Description | Required | Default
----|---|---|---
-`server` | Server must be defined in yaml and selected when using the generator | Yes | -
-`user` | User for the server to generate code for. This can also be provided as an environment variable (see below) | No | app
-`password` | Password for the server to generate code for. This can also be provided as an environment variable (see below) | No | passw0rd
-`package` | Java package name for generated code | No | com.asyncapi
-`mqTopicPrefix` | MQ topic prefix. Used for ibmmq protocols. Default will work with dev MQ instance | No | dev//
-`asyncapiFileDir` | Custom output location of the AsyncAPI file that you provided as an input | No | The root of the output directory
-
+| Name              | Description                                                                                                    | Required | Default                          |
+|-------------------|----------------------------------------------------------------------------------------------------------------|----------|----------------------------------|
+| `server`          | Server must be defined in yaml and selected when using the generator                                           | Yes      | -                                |
+| `library`         | The library to use. Current supported options: [`java`, `spring`]                                              | No       | java                             |
+| `user`            | User for the server to generate code for. This can also be provided as an environment variable (see below)     | No       | app                              |
+| `password`        | Password for the server to generate code for. This can also be provided as an environment variable (see below) | No       | passw0rd                         |
+| `package`         | Java package name for generated code                                                                           | No       | com.asyncapi                     |
+| `mqTopicPrefix`   | MQ topic prefix. Used for ibmmq protocols. Default will work with dev MQ instance                              | No       | dev//                            |
+| `asyncapiFileDir` | Custom output location of the AsyncAPI file that you provided as an input                                      | No       | The root of the output directory |
 
 ## Environment variables
 

components/Connection/NoOpConnection.js

@@ -0,0 +1,19 @@
+/*
+* (c) Copyright IBM Corporation 2021
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+export function Connection() {
+  return null;
+}

components/Connection/index.js

@@ -1,23 +1,31 @@
 import { Connection as KafkaConnection } from './KafkaConnection';
 import { Connection as MQConnection } from './MQConnection';
+import { Connection as NoOpConnection } from './NoOpConnection';
 
 const connectionModuleMap = [
   {
     protocols: ['ibmmq', 'ibmmq-secure'],
+    libraries: ['java'],
     module: MQConnection
   },
   {
     protocols: ['kafka', 'kafka-secure'],
+    libraries: ['java'],
     module: KafkaConnection
-  }
+  },
+  {
+    protocols: ['ibmmq', 'ibmmq-secure', 'kafka', 'kafka-secure'],
+    libraries: ['spring'],
+    module: NoOpConnection
+  },
 ];
 
 export default function({ asyncapi, params }) {
   const server = asyncapi.allServers().get(params.server);
   const protocol = server.protocol();
-  const foundModule = connectionModuleMap.find(item => item.protocols.includes(protocol));
+  const foundModule = connectionModuleMap.find(item => item.protocols.includes(protocol) && item.libraries.includes(params.library));
   if (!foundModule) {
-    throw new Error(`This template does not currently support the protocol ${protocol}`);
+    throw new Error(`This template does not currently support the protocol ${protocol} and library ${params.library}`);
   }
   return foundModule.module();
 }

components/ConnectionHelper/NoOpConnectionHelper.js

@@ -0,0 +1,19 @@
+/*
+* (c) Copyright IBM Corporation 2021
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+export function ConnectionHelper({ params }) {
+  return null;
+}

components/ConnectionHelper/index.js

@@ -1,23 +1,31 @@
 import { ConnectionHelper as KafkaConnectionHelper } from './KafkaConnectionHelper';
 import { ConnectionHelper as MQConnectionHelper } from './MQConnectionHelper';
+import { ConnectionHelper as NoOpConnectionHelper } from './NoOpConnectionHelper';
 
 const connectionModuleMap = [
   {
     protocols: ['ibmmq', 'ibmmq-secure'],
+    libraries: ['java'],
     module: MQConnectionHelper
   },
   {
     protocols: ['kafka', 'kafka-secure'],
+    libraries: ['java'],
     module: KafkaConnectionHelper
+  },
+  {
+    protocols: ['ibmmq', 'ibmmq-secure', 'kafka', 'kafka-secure'],
+    libraries: ['spring'],
+    module: NoOpConnectionHelper
   }
 ];
 
 export default function({ asyncapi, params }) {
   const server = asyncapi.allServers().get(params.server);
   const protocol = server.protocol();
-  const foundModule = connectionModuleMap.find(item => item.protocols.includes(protocol));
+  const foundModule = connectionModuleMap.find(item => item.protocols.includes(protocol) && item.libraries.includes(params.library));
   if (!foundModule) {
-    throw new Error(`This template does not currently support the protocol ${protocol}`);
+    throw new Error(`This template does not currently support the protocol ${protocol} and library ${params.library}`);
   }
   return foundModule.module({ asyncapi, params });
 }

components/Consumer/SpringKafkaConsumer.js

@@ -0,0 +1,59 @@
+/*
+* (c) Copyright IBM Corporation 2021
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+import { toJavaClassName } from '../../utils/String.utils';
+
+export function ConsumerDeclaration() {
+  return `
+  private final String channelName;
+  `;
+}
+
+export function ConsumerImports({ params, message }) {
+  const id = message.id() || message.name();
+  return `
+import ${params.package}.PubSubBase;
+import ${params.package}.models.ModelContract;
+import ${params.package}.models.${toJavaClassName(id)};
+
+import org.springframework.kafka.annotation.KafkaConsumer;
+`;
+}
+
+export function ReceiveMessage({ message }) {
+  const id = message.id() || message.name();
+  let groupId = '';
+  if (message.groupId) {
+    groupId = `, groupId = "${message.groupId}"`;
+  }
+  return `
+  @KafkaConsumer(topics = channelName ${groupId})
+  public void receive(${toJavaClassName(id)} ${id}) {
+    logger.info("Received message type: " + receivedObject.getClass().getName());
+    }
+  }
+  `;
+}
+
+export function ConsumerConstructor({ name }) {
+  return `
+    super();
+    this.channelName = "${name}";
+  `;
+}
+
+export function ConsumerClose() {
+  return '';
+}

components/Consumer/index.js

@@ -1,23 +1,31 @@
 import * as MQConsumer from './MQConsumer';
 import * as KafkaConsumer from './KafkaConsumer';
+import * as SpringKafkaConsumer from './SpringKafkaConsumer';
 
 const consumerModuleMap = [
   {
     protocols: ['ibmmq', 'ibmmq-secure'],
+    libraries: ['java'],
     module: MQConsumer
   },
   {
     protocols: ['kafka', 'kafka-secure'],
+    libraries: ['java'],
     module: KafkaConsumer
+  },
+  {
+    protocols: ['kafka', 'kafka-secure'],
+    libraries: ['spring'],
+    module: SpringKafkaConsumer
   }
 ];
 
 function getModule({ asyncapi, params }) {
   const server = asyncapi.allServers().get(params.server);
   const protocol = server.protocol();
-  const foundModule = consumerModuleMap.find(item => item.protocols.includes(protocol));
+  const foundModule = consumerModuleMap.find(item => item.protocols.includes(protocol) && item.libraries.includes(params.library));
   if (!foundModule) {
-    throw new Error(`This template does not currently support the protocol ${protocol}`);
+    throw new Error(`This template does not currently support the protocol ${protocol} and library ${params.library}`);
   }
   return foundModule.module;
 }

components/Demo/Demo.js

@@ -23,6 +23,9 @@ import { getMessagePayload } from '../../utils/Models.utils';
 import { PackageDeclaration } from '../Common';
 
 export function Demo(asyncapi, params) {
+  if (params.library === 'spring') {
+    return null;
+  }
   const foundPubAndSub = asyncapi.allChannels().filterBy((chan) => {
     return chan.operations().filterBySend().length > 0 &&
            chan.operations().filterByReceive().length > 0;

components/Files/Consumers.js

@@ -41,7 +41,7 @@ export function Consumers(asyncapi, channels, params) {
               <ConsumerConstructor asyncapi={asyncapi} params={params} name={name}/>
             </ClassConstructor>
 
-            <ReceiveMessage asyncapi={asyncapi} params={params} message={message}></ReceiveMessage>
+            <ReceiveMessage asyncapi={asyncapi} params={params} name={name} message={message}></ReceiveMessage>
 
             <ConsumerClose asyncapi={asyncapi} params={params} />
           </Class>

components/Files/Producers.js

@@ -25,6 +25,7 @@ export function Producers(asyncapi, channels, params) {
       const name = channel.id();
       const className = `${toJavaClassName(name)}Producer`;
       const packagePath = javaPackageToPath(params.package);
+      const classToExtend = params.library === 'spring' ? undefined : 'PubSubBase';
 
       return (
         <File name={`${packagePath}${className}.java`}>
@@ -33,7 +34,7 @@ export function Producers(asyncapi, channels, params) {
           <ProducerImports asyncapi={asyncapi} params={params} />
           <ImportModels asyncapi={asyncapi} params={params} />
 
-          <Class name={className} extendsClass="PubSubBase">
+          <Class name={className} extendsClass={classToExtend}>
             <ProducerDeclaration asyncapi={asyncapi} params={params} />
 
             <ClassConstructor name={className}>

components/PomHelper.js

@@ -21,7 +21,8 @@ import { render } from '@asyncapi/generator-react-sdk';
 export function PomHelper({ server, params }) {
   // Resolve additional dependencies depending on protocol supplied
   const supportedProtocol = server.protocol();
-  const dependencies = resolveDependencies(supportedProtocol);
+  const library = params.library;
+  const dependencies = resolveDependencies(supportedProtocol, library);
 
   let protocolDependencies = '';
 

components/Producer/SpringKafkaProducer.js

@@ -0,0 +1,85 @@
+/*
+* (c) Copyright IBM Corporation 2021
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+export function ProducerDeclaration() {
+  return `
+    private SpringKafkaProducer producer = null;
+  `;
+}
+
+export function SendMessage() {
+  return `
+    public void send(ModelContract modelContract) {
+        Serializable modelInstance = (Serializable) modelContract;
+
+        try{
+          // JSON encode and transmit
+          ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter();
+          String json = ow.writeValueAsString(modelInstance);
+
+          logger.info("Sending Message: " + json);
+
+          ProducerRecord<String, String> record = new ProducerRecord<String, String>(topicName, json);
+          producer.send(record);
+
+        }catch (Exception e){
+          logger.severe("An error occured whilst attempting to send a message: " + e);
+        }
+    }`;
+}
+
+export function ProducerConstructor({ name }) {
+  return `
+    super();
+    String id = "my-publisher";
+
+    logger.info("Pub application is starting");
+
+    // prepare connection for producer
+    createConnection("${name}", id);
+
+    producer = ch.createProducer();
+`;
+}
+
+export function ProducerClose() {
+  return `
+    public void close() {
+      producer.close();
+    }
+  `;
+}
+
+export function ProducerImports({ params }) {
+  return `
+import java.util.logging.*;
+import java.io.Serializable;
+import java.util.UUID;
+
+import ${params.package}.ConnectionHelper;
+import ${params.package}.LoggingHelper;
+import ${params.package}.Connection;
+import ${params.package}.PubSubBase;
+import ${params.package}.models.ModelContract;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.annotation.JsonView;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+  `;
+}

components/Producer/index.js

@@ -1,23 +1,31 @@
 import * as MQProducer from './MQProducer';
 import * as KafkaProducer from './KafkaProducer';
+import * as SpringKafkaProducer from './SpringKafkaProducer';
 
 const producerModuleMap = [
   {
     protocols: ['ibmmq', 'ibmmq-secure'],
+    libraries: ['java'],
     module: MQProducer
   },
   {
     protocols: ['kafka', 'kafka-secure'],
+    libraries: ['java'],
     module: KafkaProducer
+  },
+  {
+    protocols: ['kafka', 'kafka-secure'],
+    libraries: ['spring'],
+    module: SpringKafkaProducer
   }
 ];
 
 function getModule({ asyncapi, params }) {
   const server = asyncapi.allServers().get(params.server);
   const protocol = server.protocol();
-  const foundModule = producerModuleMap.find(item => item.protocols.includes(protocol));
+  const foundModule = producerModuleMap.find(item => item.protocols.includes(protocol) && item.libraries.includes(params.library));
   if (!foundModule) {
-    throw new Error(`This template does not currently support the protocol ${protocol}`);
+    throw new Error(`This template does not currently support the protocol ${protocol} and library ${params.library}`);
   }
   return foundModule.module;
 }