<commit_before>package io.scalecube.services.codec;

import io.scalecube.services.api.ErrorData;
import io.scalecube.services.api.ServiceMessage;
import io.scalecube.services.exceptions.BadRequestException;
import io.scalecube.services.exceptions.ExceptionProcessor;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.util.ReferenceCountUtil;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.Optional;

public final class ServiceMessageDataCodec {

  private static final Logger LOGGER = LoggerFactory.getLogger(ServiceMessageDataCodec.class);

  private static final String DEFAULT_DATA_FORMAT = "application/json";

  public ServiceMessage encode(ServiceMessage message) {
    if (message.hasData(ByteBuf.class)) {
      return message;
    } else if (message.hasData()) {
      ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
      try {
        String contentType = Optional.ofNullable(message.dataFormat()).orElse(DEFAULT_DATA_FORMAT);
        DataCodec dataCodec = DataCodec.getInstance(contentType);
        dataCodec.encode(new ByteBufOutputStream(buffer), message.data());
        return ServiceMessage.from(message).data(buffer).build();
      } catch (Throwable ex) {
        ReferenceCountUtil.release(buffer);
        LOGGER.error("Failed to encode data on: {}, cause: {}", message, ex);
        throw new BadRequestException("Failed to encode data on message q=" + message.qualifier());
      }
    }
    return message;
  }

  public ServiceMessage decode(ServiceMessage message, Class<?> type) {
    if (!message.hasData(ByteBuf.class) || type == null) {
      return message;
    }

    Object data;
    Class<?> targetType = ExceptionProcessor.isError(message) ? ErrorData.class : type;

    try (ByteBufInputStream inputStream = new ByteBufInputStream(((ByteBuf) message.data()).slice())) {
      String contentType = Optional.ofNullable(message.dataFormat()).orElse(DEFAULT_DATA_FORMAT);
      DataCodec dataCodec = DataCodec.getInstance(contentType);
      data = dataCodec.decode(inputStream, targetType);
    } catch (Throwable ex) {
      LOGGER.error("Failed to decode data on: {}, data buffer: {}, cause: {}",
          message, ex, ((ByteBuf) message.data()).toString(Charset.defaultCharset()));
      throw new BadRequestException("Failed to decode data on message q=" + message.qualifier());
    } finally {
      ReferenceCountUtil.release(message.data());
    }

    if (targetType == ErrorData.class) {
      throw ExceptionProcessor.toException(message.qualifier(), (ErrorData) data);
    }
    return ServiceMessage.from(message).data(data).build();
  }
}<commit_msg>Minor<commit_after>package io.scalecube.services.codec;

import io.scalecube.services.api.ErrorData;
import io.scalecube.services.api.ServiceMessage;
import io.scalecube.services.exceptions.BadRequestException;
import io.scalecube.services.exceptions.ExceptionProcessor;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.util.ReferenceCountUtil;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.Optional;

public final class ServiceMessageDataCodec {

  private static final Logger LOGGER = LoggerFactory.getLogger(ServiceMessageDataCodec.class);

  private static final String DEFAULT_DATA_FORMAT = "application/json";

  public ServiceMessage encode(ServiceMessage message) {
    if (message.hasData(ByteBuf.class)) {
      return message;
    } else if (message.hasData()) {
      ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
      try {
        String contentType = Optional.ofNullable(message.dataFormat()).orElse(DEFAULT_DATA_FORMAT);
        DataCodec dataCodec = DataCodec.getInstance(contentType);
        dataCodec.encode(new ByteBufOutputStream(buffer), message.data());
        return ServiceMessage.from(message).data(buffer).build();
      } catch (Throwable ex) {
        ReferenceCountUtil.release(buffer);
        LOGGER.error("Failed to encode data on: {}, cause: {}", message, ex);
        throw new BadRequestException("Failed to encode data on message q=" + message.qualifier());
      }
    }
    return message;
  }

  public ServiceMessage decode(ServiceMessage message, Class<?> type) {
    if (!message.hasData(ByteBuf.class) || type == null) {
      return message;
    }

    Object data;
    Class<?> targetType = ExceptionProcessor.isError(message) ? ErrorData.class : type;

    ByteBuf dataBuffer = message.data();
    try (ByteBufInputStream inputStream = new ByteBufInputStream(dataBuffer.slice())) {
      String contentType = Optional.ofNullable(message.dataFormat()).orElse(DEFAULT_DATA_FORMAT);
      DataCodec dataCodec = DataCodec.getInstance(contentType);
      data = dataCodec.decode(inputStream, targetType);
    } catch (Throwable ex) {
      LOGGER.error("Failed to decode data on: {}, cause: {}, data buffer: {}",
          message, ex, dataBuffer.toString(Charset.defaultCharset()));
      throw new BadRequestException("Failed to decode data on message q=" + message.qualifier());
    } finally {
      ReferenceCountUtil.release(dataBuffer);
    }

    if (targetType == ErrorData.class) {
      throw ExceptionProcessor.toException(message.qualifier(), (ErrorData) data);
    }
    return ServiceMessage.from(message).data(data).build();
  }
}
{'fraction_non_alphanumeric': 0.11507438609069726, 'fraction_numerical': 0.000716974368166338, 'mean_word_length': 4.299145299145299, 'pattern_counts': {'":': 0, '<': 7, '<?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': '6038606', 'n_tokens_mistral': 1545, 'n_tokens_neox': 1557, 'n_words': 329}