D-Net

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

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

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<>();

		}

		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<>();

			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, (JsonDeserializer<VertexWritable>) (json1, typeOfT, context) -> {

			JsonObject jo = json1.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, (JsonSerializer<VertexWritable>) (src, typeOfSrc, 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<>();

			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;

	}

}