Use generic serializer

[pmerienne]

Hi,

I really like your project and I think I will use it very soon. It's quite similar to https://github.com/hmsonline/trident-elasticsearch but you wrote some useful documentation and your code is really cleaner!

I think you can improve your API by using a generic serializer. This way you can avoid passing the class to the ESIndexStateFactory.

e.g

TridentState staticState = topology.newStaticState(new ESIndexState.Factory<>(getLocalClient(), Tweet.class));

will be

TridentState staticState = topology.newStaticState(new ESIndexState.Factory<>(getLocalClient()));

It's not a major improvement but I was quite happy when I added it to a similar project for hbase (not in github). Do achieve this, you need to use fastjson library. Quick draft :

public class JsonValueSerializer<T> {

    public byte[] serialize(T obj) {
        try {
            return JSON.toJSONString(obj, SerializerFeature.WriteClassName).getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }

    public T deserialize(byte[] bytes) {
        try {
            String json = new String(bytes, "UTF-8");
            return (T) JSON.parse(json);
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }
}

Then you can create Transactional/Opaque serializer like :

 public class TransactionalSerializer {

    private final JsonValueSerializer delegate;

    public TransactionalSerializer(JsonValueSerializer delegate) {
        this.delegate = delegate;
    }

    public byte[] serialize(TransactionalValue obj) {
        List<Object> toSer = new ArrayList<>(2);
        toSer.add(obj.getTxid());
        toSer.add(obj.getVal());
        return delegate.serialize(toSer);
    }

    public TransactionalValue deserialize(byte[] b) {
        List deser = (List) delegate.deserialize(b);
        return new TransactionalValue((Long) deser.get(0), deser.get(1));
    }

}