TestRabbitConsumerJob.java

package com.squarespace.hamster.rabbitmq.v6;

import com.squarespace.hamster.runtime.client.Context;
import com.squarespace.hamster.runtime.client.Job;
import com.squarespace.hamster.runtime.client.serialization.Serde;
import com.squarespace.v6.model.trashcan.queue.TrashcanDeletionMessage;

public class TestRabbitConsumerJob implements Job<TrashcanDeletionMessage, TrashcanDeletionMessage> {
  private V6MessageSerde<TrashcanDeletionMessage, TrashcanDeletionMessage> serde;

  public TestRabbitConsumerJob() {
    serde = new V6MessageSerde<>(TrashcanDeletionMessage.class);
  }


  @Override
  public TrashcanDeletionMessage execute(TrashcanDeletionMessage trashcanDeletionMessageMessage, Context context) throws Exception {
    return null;
  }

  @Override
  public Serde<TrashcanDeletionMessage, TrashcanDeletionMessage> getSerde() {
    return serde;
  }
}

V6MessageSerde.java

package com.squarespace.hamster.rabbitmq.v6;

import java.io.IOException;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.squarespace.hamster.runtime.client.serialization.Deserializer;
import com.squarespace.hamster.runtime.client.serialization.Serde;
import com.squarespace.hamster.runtime.client.serialization.Serializer;
import com.squarespace.util.rabbitmq.message.Data;
import com.squarespace.util.rabbitmq.message.Message;
import com.squarespace.v6.utils.JSONInternalMapper;

public class V6MessageSerde<I extends Data, O extends Data> implements Serde<I, O> {
  private V6ObjectSerializer<O> serializer;
  private V6ObjectDeserializer<I> deserializer;
  private ObjectMapper objectMapper;

  public V6MessageSerde(Class<I> inputType) {
    objectMapper = JSONInternalMapper.getMapper();
    serializer = new V6ObjectSerializer<>(objectMapper);
    deserializer = new V6ObjectDeserializer<>(objectMapper, inputType);
  }

  @Override
  public Serializer<O> serializer() {
    return serializer;
  }

  @Override
  public Deserializer<I> deserializer() {
    return deserializer;
  }


  private static class V6ObjectSerializer<T extends Data> implements Serializer<T> {
    private ObjectMapper objectMapper;

    public V6ObjectSerializer(ObjectMapper objectMapper) {
      this.objectMapper = objectMapper;
    }

    @Override
    public byte[] serialize(T t) {
      // create Message instance and serialize
      return null;
    }
  }

  private static class V6ObjectDeserializer<T extends Data> implements Deserializer<T> {
    private ObjectMapper objectMapper;
    private Class<T> dataType;

    public V6ObjectDeserializer(ObjectMapper objectMapper, Class<T> dataType) {
      this.objectMapper = objectMapper;
      this.dataType = dataType;
    }

    @Override
    public T deserialize(byte[] bytes) {
      try {
        Message<T> message = (Message<T>) objectMapper.readValue(bytes, Message.class);

        // Check the type.
        T data = message.getData();
        if (data != null && !dataType.isInstance(data)) {
          throw new ClassCastException(dataType + " vs. " + data.getClass());
        }

        return data;
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  }
}