code: /*******************************************************************************
   * Copyright 2018 572682
   * 
   * 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.
   ******************************************************************************/
  package us.dot.its.jpo.ode.services.asn1;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Controller;
  
  import us.dot.its.jpo.ode.OdeProperties;
  import us.dot.its.jpo.ode.wrapper.MessageConsumer;
  
  /**
   * Launches ToJsonConverter service
   */
  @Controller
  public class AsnCodecRouterServiceController {
  
     private static final Logger logger = LoggerFactory.getLogger(AsnCodecRouterServiceController.class);
     org.apache.kafka.common.serialization.Serdes bas;
  
     @Autowired
     public AsnCodecRouterServiceController(OdeProperties odeProps) {
        super();
  
        logger.info("Starting {}", this.getClass().getSimpleName());
  
        // asn1_codec Decoder Routing
        logger.info("Routing DECODED data received ASN.1 Decoder");
  
        Asn1DecodedDataRouter decoderRouter = new Asn1DecodedDataRouter(odeProps);
  
        MessageConsumer<String, String> asn1DecoderConsumer = MessageConsumer.defaultStringMessageConsumer(
           odeProps.getKafkaBrokers(), this.getClass().getSimpleName(), decoderRouter);
  
        asn1DecoderConsumer.setName("Asn1DecoderConsumer");
        decoderRouter.start(asn1DecoderConsumer, odeProps.getKafkaTopicAsn1DecoderOutput());
  
        // asn1_codec Encoder Routing
        logger.info("Routing ENCODED data received ASN.1 Encoder");
  
        Asn1EncodedDataRouter encoderRouter = new Asn1EncodedDataRouter(odeProps);
  
        MessageConsumer<String, String> encoderConsumer = MessageConsumer.defaultStringMessageConsumer(
           odeProps.getKafkaBrokers(), this.getClass().getSimpleName(), encoderRouter);
  
        encoderConsumer.setName("Asn1EncoderConsumer");
        encoderRouter.start(encoderConsumer, odeProps.getKafkaTopicAsn1EncoderOutput());
     }
  }
  