D-Net

package eu.dnetlib.data.mapreduce.hbase.dedup.cc;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import java.lang.reflect.Type;

import java.util.TreeSet;

import com.google.gson.*;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.Writable;

/**

 * Created by claudio on 14/10/15.

 */

public final class VertexWritable implements Writable, Cloneable {

	private Text vertexId;

	private TreeSet<Text> edges;

	private boolean activated;

	public VertexWritable() {

		super();

	}

	public VertexWritable(Text minimalVertexId) {

		super();

		this.vertexId = minimalVertexId;

	}

	// true if updated

	public boolean checkAndSetMinimalVertex(Text id) {

		if (vertexId == null) {

			vertexId = id;

			return true;

		} else {

			if (id.compareTo(vertexId) < 0) {

				vertexId = id;

				return true;

			}

		}

		return false;

	}

	public boolean isMessage() {

		return (edges == null);

	}

	public VertexWritable makeMessage() {

		return new VertexWritable(vertexId);

	}

	public void addVertex(Text id) {

		if (edges == null) {

			edges = new TreeSet<Text>();

		}

		edges.add(id);

	}

	@Override

	public void write(DataOutput out) throws IOException {

		vertexId.write(out);

		if (edges == null) {

			out.writeInt(-1);

		} else {

			out.writeInt(edges.size());

			for (Text l : edges) {

				l.write(out);

			}

		}

		out.writeBoolean(activated);

	}

	@Override

	public void readFields(DataInput in) throws IOException {

		vertexId = new Text();

		vertexId.readFields(in);

		int length = in.readInt();

		if (length > -1) {

			edges = new TreeSet<Text>();

			for (int i = 0; i < length; i++) {

				Text temp = new Text();

				temp.readFields(in);

				edges.add(temp);

			}

		} else {

			edges = null;

		}

		activated = in.readBoolean();

	}

	@Override

	public String toString() {

		StringBuilder sb = new StringBuilder();

		sb.append("VertexWritable [minimalVertexId=").

		append(vertexId).

		append(", pointsTo=").

		append(edges).

		append("]");

		return sb.toString();

	}

	public static VertexWritable fromJSON(String json) {

		final GsonBuilder builder = new GsonBuilder();

		builder.registerTypeAdapter(VertexWritable.class, new JsonDeserializer<VertexWritable>() {

			@Override

			public VertexWritable deserialize(final JsonElement json, final Type typeOfT, final JsonDeserializationContext context) throws JsonParseException {

				JsonObject jo = json.getAsJsonObject();

				VertexWritable vertexWritable = new VertexWritable();

				vertexWritable.setVertexId(new Text(jo.get("vertexId").getAsString()));

				vertexWritable.setActivated(jo.get("activated").getAsBoolean());

				TreeSet<Text> edges = new TreeSet<Text>();

				for(JsonElement e : jo.get("edges").getAsJsonArray()) {

					edges.add(new Text(e.getAsString()));

				}

				vertexWritable.setEdges(edges);

				return vertexWritable;

			}

		});

		return builder.create().fromJson(json, VertexWritable.class);

	}

	public String toJSON() {

		final GsonBuilder builder = new GsonBuilder();

		builder.registerTypeAdapter(VertexWritable.class, new JsonSerializer<VertexWritable>() {

			@Override

			public JsonElement serialize(final VertexWritable src, final Type typeOfSrc, final JsonSerializationContext context) {

				JsonObject res = new JsonObject();

				res.addProperty("vertexId", getVertexId().toString());

				res.addProperty("activated", isActivated());

				JsonArray edges = new JsonArray();

				if (!isMessage()) {

					for (Text edge : getEdges()) {

						edges.add(new JsonPrimitive(edge.toString()));

					}

				}

				res.add("edges", edges);

				return res;

			}

		});

		return builder.create().toJson(this);

	}

	@Override

	protected VertexWritable clone() {

		VertexWritable toReturn = new VertexWritable(new Text(vertexId.toString()));

		if (edges != null) {

			toReturn.edges = new TreeSet<Text>();

			for (Text l : edges) {

				toReturn.edges.add(new Text(l.toString()));

			}

		}

		return toReturn;

	}

	public boolean isActivated() {

		return activated;

	}

	public void setActivated(boolean activated) {

		this.activated = activated;

	}

	public Text getVertexId() {

		return vertexId;

	}

	public void setVertexId(Text vertexId) {

		this.vertexId = vertexId;

	}

	public TreeSet<Text> getEdges() {

		return edges;

	}

	public void setEdges(TreeSet<Text> edges) {

		this.edges = edges;

	}

}