/*
 * Copyright 2012-2013 Continuuity,Inc. All Rights Reserved.
 */

package com.continuuity.internal.app;

import com.continuuity.api.flow.flowlet.FailurePolicy;
import com.continuuity.api.flow.flowlet.FlowletSpecification;
import com.continuuity.internal.api.flowlet.DefaultFlowletSpecification;
import com.google.gson.JsonDeserializationContext;
import com.google.gson.JsonDeserializer;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParseException;
import com.google.gson.JsonPrimitive;
import com.google.gson.JsonSerializationContext;
import com.google.gson.JsonSerializer;

import java.lang.reflect.Type;

/**
 *
 */
final class FlowletSpecificationCodec implements JsonSerializer<FlowletSpecification>,
                                                   JsonDeserializer<FlowletSpecification> {

  @Override
  public JsonElement serialize(FlowletSpecification src, Type typeOfSrc, JsonSerializationContext context) {
    JsonObject jsonObj = new JsonObject();

    jsonObj.add("className", new JsonPrimitive(src.getClassName()));
    jsonObj.add("name", new JsonPrimitive(src.getName()));
    jsonObj.add("description", new JsonPrimitive(src.getDescription()));
    jsonObj.add("failurePolicy", new JsonPrimitive(src.getFailurePolicy().name()));

    return jsonObj;
  }

  @Override
  public FlowletSpecification deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
    JsonObject jsonObj = json.getAsJsonObject();

    String className = jsonObj.get("className").getAsString();
    String name = jsonObj.get("name").getAsString();
    String description = jsonObj.get("description").getAsString();
    FailurePolicy policy = FailurePolicy.valueOf(jsonObj.get("failurePolicy").getAsString());

    return new DefaultFlowletSpecification(className, name, description, policy);
  }
}

@@ -283,32 +283,76 @@
 tSpecification;%0A
+import com.google.common.reflect.TypeToken;%0A
 import com.googl
@@ -698,16 +698,38 @@
 ct.Type;
+%0Aimport java.util.Set;
 %0A%0A/**%0A *
@@ -1362,16 +1362,125 @@
 ame()));
+%0A    jsonObj.add(%22datasets%22, context.serialize(src.getDataSets(), new TypeToken%3CSet%3CString%3E%3E()%7B%7D.getType()));
 %0A%0A    re
@@ -1974,16 +1974,131 @@
 ring());
+%0A    Set%3CString%3E dataSets = context.deserialize(jsonObj.get(%22datasets%22), new TypeToken%3CSet%3CString%3E%3E()%7B%7D.getType());
 %0A%0A    re
@@ -2170,16 +2170,26 @@
 , policy
+, dataSets
 );%0A  %7D%0A%7D

Fix the FlowletSpecificationCodec for the new DataSets property.