package pl.allegro.tech.hermes.consumers.consumer.sender.googlepubsub;

import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import pl.allegro.tech.hermes.consumers.consumer.Message;
import pl.allegro.tech.hermes.consumers.consumer.trace.MetadataAppender;

import java.io.IOException;

class GooglePubSubMessageTransformerCompression implements GooglePubSubMessageTransformer {

    private final MessageCompressor compressor;
    private final Long compressionThresholdBytes;
    private final GooglePubSubMessageTransformerRaw rawTransformer;
    private final MetadataAppender<PubsubMessage> metadataAppender;

    GooglePubSubMessageTransformerCompression(
            Long compressionThresholdBytes,
            GooglePubSubMessageTransformerRaw rawTransformer,
            MetadataAppender<PubsubMessage> metadataAppender,
            MessageCompressor compressor) {
        this.compressor = compressor;
        this.compressionThresholdBytes = compressionThresholdBytes;
        this.rawTransformer = rawTransformer;
        this.metadataAppender = metadataAppender;
    }

    @Override
    public PubsubMessage fromHermesMessage(Message message) {
        byte[] data = message.getData();
        if (data.length > compressionThresholdBytes) {
            return compressHermesMessage(message);
        } else {
            return rawTransformer.fromHermesMessage(message);
        }
    }

    private PubsubMessage compressHermesMessage(Message message) {
        try {
            final PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
                    .setData(ByteString.copyFrom(
                            compressor.compress(message.getData())
                    )).build();

            return metadataAppender.append(pubsubMessage, message);
        } catch (IOException e) {
            throw new GooglePubSubMessageCompressionException("Error on PubSub message compression", e);
        }
    }
}
