Prevent the same kafka topic name from being use both as an input and as an output topic name

webapp/frontend/app/components/kafka-topics.js

@@ -3,11 +3,13 @@ import Ember from 'ember';
 export default Ember.Component.extend({
   topics: [],
 
+  id: "",
+
   didInsertElement: function(){
     this._super(...arguments);
 
     // Initialize tokenfield.
-    this.$('#tokenfield')
+    this.$('#' + this.get('id'))
       .on('tokenfield:createtoken', this.createTokenEvent)
       .tokenfield({
         createTokensOnBlur: true
@@ -26,7 +28,14 @@ export default Ember.Component.extend({
 
   actions: {
     updateTopics: function(){
-      this.set('topics', this.$('#tokenfield').tokenfield('getTokensList').replace(/\s+/g, '').split(','));
+      var tokenList = this.$('#' + this.get('id')).tokenfield('getTokensList', ',', false, false);
+
+      if (tokenList === "") {
+        this.set('topics', []);
+      }
+      else{
+        this.set('topics', tokenList.split(','));
+      }
     }
   }
 

webapp/frontend/app/controllers/create-lambda-instance.js

@@ -46,6 +46,9 @@ export default Ember.Controller.extend({
 
   kafkaInputTopics: ["input"],
   kafkaOutputTopics: ["batch-output", "stream-output"],
+  conflictingKafkaTopics: false,
+  kafkaTopicsValidityReported: false,
+  kafkaTopicsConflictMessage: "Using the same name for input and output topic will result only in input topic creation!",
 
   actions: {
     saveLambdaInstance: function(newLambdaInstance){
@@ -301,15 +304,53 @@ export default Ember.Controller.extend({
   },
 
   kafkaInputTopicsObserver: Ember.observer('kafkaInputTopics', function() {
-    this.parseKafkaTopics("Input");
+    this.parseKafkaTopics();
   }),
 
   kafkaOutputTopicsObserver: Ember.observer('kafkaOutputTopics', function() {
-    this.parseKafkaTopics("Output");
+    this.parseKafkaTopics();
   }),
 
-  parseKafkaTopics: function(message){
-    console.log(message);
+  parseKafkaTopics: function(){
+    // In case this method has been called due to a change in kafkaInputTopics or kafkaOutputTopics
+    // from inside the route, then these elements will be undefined since the HTML hasn't yet been
+    // rendered.
+    var inputTopicsElement = Ember.$('#kafka-input-topics #kafka-input-topics-tokenfield')[0];
+    var outputTopicsElement = Ember.$('#kafka-output-topics #kafka-output-topics-tokenfield')[0];
+
+    var kafkaInputTopics = this.get('kafkaInputTopics');
+    var kafkaOutputTopics = this.get('kafkaOutputTopics');
+
+    for (var i = 0, n = kafkaInputTopics.get('length');i < n;i++) {
+      for (var j = 0, m = kafkaOutputTopics.get('length'); j < m;j++) {
+        if (kafkaInputTopics.objectAt(i) === kafkaOutputTopics.objectAt(j)) {
+          this.set('conflictingKafkaTopics', true);
+
+          if(!this.get('kafkaTopicsValidityReported')){
+            if(inputTopicsElement !== undefined && outputTopicsElement !== undefined){
+              inputTopicsElement.setCustomValidity(this.get('kafkaTopicsConflictMessage'));
+              outputTopicsElement.setCustomValidity(this.get('kafkaTopicsConflictMessage'));
+              inputTopicsElement.reportValidity();
+              outputTopicsElement.reportValidity();
+
+              this.set('kafkaTopicsValidityReported', true);
+            }
+          }
+
+          return;
+        }
+      }
+    }
+
+    this.set('conflictingKafkaTopics', false);
+    if(this.get('kafkaTopicsValidityReported')){
+      if(inputTopicsElement !== undefined && outputTopicsElement !== undefined){
+        inputTopicsElement.setCustomValidity("");
+        outputTopicsElement.setCustomValidity("");
+
+        this.set('kafkaTopicsValidityReported', false);
+      }
+    }
   }
 
 });

webapp/frontend/app/routes/create-lambda-instance.js

@@ -114,6 +114,8 @@ export default Ember.Route.extend(AuthenticatedRouteMixin, {
 
       controller.set('kafkaInputTopics', ["input"]);
       controller.set('kafkaOutputTopics', ["batch-output", "stream-output"]);
+      controller.set('conflictingKafkaTopics', false);
+      controller.set('kafkaTopicsValidityReported', false);
     }
   }
 });

webapp/frontend/app/templates/components/kafka-topics.hbs

-<input type="text" class="form-control" id="tokenfield" placeholder="e.g. topic1, topic2, topic3" onChange={{action "updateTopics"}}>
\ No newline at end of file
+<input type="text" class="form-control" id="{{id}}" placeholder="e.g. topic1, topic2, topic3" onChange={{action "updateTopics"}}>
\ No newline at end of file

webapp/frontend/app/templates/create-lambda-instance.hbs

@@ -97,15 +97,15 @@
 
             <style>.tokenfield.form-control { height: auto; }</style>
             <div class="row">
-              <div class="form-group col-sm-6">
+              <div class="form-group col-sm-6 {{if conflictingKafkaTopics 'has-warning'}}">
                 <label for="kafka_topics">Apache Kafka input topics:</label>
-                  {{kafka-topics topics=kafkaInputTopics}}
+                  {{kafka-topics topics=kafkaInputTopics id="kafka-input-topics"}}
                   <span id="helpBlock" class="help-block">Names of Apache Kafka input topics<br>
                   Default input topic, if no names are given: "input"</span>
               </div>
-              <div class="form-group col-sm-6 pull-right">
+              <div class="form-group col-sm-6 pull-right {{if conflictingKafkaTopics 'has-warning'}}">
                 <label for="kafka_topics">Apache Kafka output topics:</label>
-                  {{kafka-topics topics=kafkaOutputTopics}}
+                  {{kafka-topics topics=kafkaOutputTopics id="kafka-output-topics"}}
                   <span id="helpBlock" class="help-block">Names of Apache Kafka output topics<br>
                   Default output topics, if no names are given: "batch-output", "stream-output"</span>
               </div>