<commit_before>package us.dot.its.jpo.ode.udp.vsd;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.SerializationUtils;

import com.oss.asn1.EncodeFailedException;
import com.oss.asn1.EncodeNotSupportedException;

import us.dot.its.jpo.ode.OdeProperties;
import us.dot.its.jpo.ode.dds.AbstractSubscriberDepositor;
import us.dot.its.jpo.ode.j2735.semi.SemiDialogID;
import us.dot.its.jpo.ode.j2735.semi.VehSitDataMessage;
import us.dot.its.jpo.ode.udp.UdpUtil.UdpUtilException;
import us.dot.its.jpo.ode.wrapper.MessageConsumer;

/**
 * Publishes VSDs to SDC.
 */
public class VsdDepositor extends AbstractSubscriberDepositor<String, byte[]> {

   public VsdDepositor(OdeProperties odeProps) {
      super(odeProps, odeProps.getVsdDepositorPort());
      
      if (!odeProps.getDepositSanitizedBsmToSdc()) {
         logger.warn("WARNING - SDC BSM/VSD deposit option disabled, not starting VSD depositor service.");
         return;
      }
      
      consumer = MessageConsumer.defaultByteArrayMessageConsumer(odeProps.getKafkaBrokers(),
            odeProps.getHostId() + this.getClass().getSimpleName(), this);
      consumer.setName(this.getClass().getSimpleName());
   }

   @Override
   public byte[] call() {
      VehSitDataMessage vsd = (VehSitDataMessage) SerializationUtils.deserialize(record.value());
      logger.info("Received VSD.");

      byte[] encodedVsd = null;
      try {
         if (trustSession.establishTrust(vsd.requestID, SemiDialogID.vehSitData)) {
            logger.debug("Sending VSD to SDC IP: {} Port: {}", odeProperties.getSdcIp(), odeProperties.getSdcPort());

            encodedVsd = coder.encode(vsd).array();
            sendToSdc(encodedVsd);
            messagesSent++;
         } else {
            logger.error("Failed to establish trust, not sending VSD.");
         }
      } catch (UdpUtilException | EncodeFailedException | EncodeNotSupportedException e) {
         logger.error("Error Sending Isd to SDC", e);
         return new byte[0];
      }

      logger.info("VSDs sent since session start: {}/{}", messagesSent,
            odeProperties.getMessagesUntilTrustReestablished());

      if (messagesSent >= odeProperties.getMessagesUntilTrustReestablished()) {
         trustSession.setTrustEstablished(false);
      }

      return encodedVsd;
   }

   @Override
   public Logger getLogger() {
      return LoggerFactory.getLogger(this.getClass());
   }
}<commit_msg>ODE-314 bugfix<commit_after>package us.dot.its.jpo.ode.udp.vsd;

import java.io.ByteArrayInputStream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.SerializationUtils;

import com.oss.asn1.DecodeFailedException;
import com.oss.asn1.DecodeNotSupportedException;
import com.oss.asn1.EncodeFailedException;
import com.oss.asn1.EncodeNotSupportedException;

import us.dot.its.jpo.ode.OdeProperties;
import us.dot.its.jpo.ode.dds.AbstractSubscriberDepositor;
import us.dot.its.jpo.ode.j2735.J2735;
import us.dot.its.jpo.ode.j2735.dsrc.TemporaryID;
import us.dot.its.jpo.ode.j2735.semi.SemiDialogID;
import us.dot.its.jpo.ode.j2735.semi.VehSitDataMessage;
import us.dot.its.jpo.ode.udp.UdpUtil.UdpUtilException;
import us.dot.its.jpo.ode.wrapper.MessageConsumer;

/**
 * Publishes VSDs to SDC.
 */
public class VsdDepositor extends AbstractSubscriberDepositor<String, byte[]> {

   public VsdDepositor(OdeProperties odeProps) {
      super(odeProps, odeProps.getVsdDepositorPort());

      if (!odeProps.getDepositSanitizedBsmToSdc()) {
         logger.warn("WARNING - SDC BSM/VSD deposit option disabled, not starting VSD depositor service.");
         return;
      }

      consumer = MessageConsumer.defaultByteArrayMessageConsumer(odeProps.getKafkaBrokers(),
            odeProps.getHostId() + this.getClass().getSimpleName(), this);
      consumer.setName(this.getClass().getSimpleName());
   }

   @Override
   public byte[] call() {
      VehSitDataMessage vsd = (VehSitDataMessage) SerializationUtils.deserialize(record.value());
      logger.info("Received VSD.");

      byte[] encodedVsd = null;
      try {
         if (trustSession.establishTrust(vsd.requestID, SemiDialogID.vehSitData)) {
            logger.debug("Sending VSD to SDC IP: {} Port: {}", odeProperties.getSdcIp(), odeProperties.getSdcPort());

            encodedVsd = coder.encode(vsd).array();
            sendToSdc(encodedVsd);
            messagesSent++;
         } else {
            logger.error("Failed to establish trust, not sending VSD.");
         }
      } catch (UdpUtilException | EncodeFailedException | EncodeNotSupportedException e) {
         logger.error("Error Sending Isd to SDC", e);
         return new byte[0];
      }

      logger.info("VSDs sent since session start: {}/{}", messagesSent,
            odeProperties.getMessagesUntilTrustReestablished());

      if (messagesSent >= odeProperties.getMessagesUntilTrustReestablished()) {
         trustSession.setTrustEstablished(false);
      }

      return encodedVsd;
   }

   public TemporaryID getRequestId() {
      TemporaryID reqID = null;
      try {
         reqID = ((VehSitDataMessage) J2735.getPERUnalignedCoder().decode(new ByteArrayInputStream(record.value()),
               new VehSitDataMessage())).requestID;

      } catch (DecodeFailedException | DecodeNotSupportedException e) {
         logger.error("Depositor failed to decode ISD message: {}", e);
      }

      return reqID;
   }

   @Override
   public Logger getLogger() {
      return LoggerFactory.getLogger(this.getClass());
   }
}
{'fraction_non_alphanumeric': 0.11667564171602944, 'fraction_numerical': 0.008436546401005206, 'mean_word_length': 3.4152139461172744, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7322443', 'n_tokens_mistral': 1802, 'n_tokens_neox': 1734, 'n_words': 295}