public static JSONArtifact string2JSON(
	String data
   ) {
	if( data == null || data.length() == 0 )
	{
		return null;
	}
	
   	JSONArtifact jArtifact;
   	try
   	{
       	jArtifact = JSON.parse(  data );
   	}
   	catch( Throwable t )
   	{
   		return null;	// Not JSON
   	}
	return jArtifact;
}
 

/**
 * Convert a JSON stream to an SPLStream.
 * @param stream JSON stream to be converted.
 * @return SPLStream with schema {@link JSONSchemas#JSON}.
 */
public static SPLStream toSPL(TStream<JSONObject> stream) {
    
    return SPLStreams.convertStream(stream, 
            new BiFunction<JSONObject, OutputTuple, OutputTuple>() {
                private static final long serialVersionUID = 1L;

                @Override
                public OutputTuple apply(JSONObject v1, OutputTuple v2) {
                    try {
                        v2.setString(0, v1.serialize());
                        return v2;
                    } catch (IOException e) {
                        return null;
                    }
                }
    }, JSONSchemas.JSON);
}
 

private void checkJsonOutput(JSONArtifact expected,
          TStream<String> jsonString) throws Exception, IOException {
             
      assertEquals(String.class, jsonString.getTupleClass());
      assertEquals(String.class, jsonString.getTupleType());
             
      Condition<?> valueChecker = jsonString.topology().getTester().stringTupleTester(jsonString, v -> {
	try {
		return expected.equals(JSON.parse(v));
	} catch (IOException e) {
		throw new RuntimeException(e);
	}
});

      Condition<Long> singleTuple = jsonString.topology().getTester().tupleCount(jsonString, 1);
      complete(jsonString.topology().getTester(), singleTuple.and(valueChecker), 10, TimeUnit.SECONDS);
      
      assertTrue(singleTuple.valid());
      assertTrue(valueChecker.valid());
  }
 

/**
 * Test that if the serialized value is
 * an array, it ends up wrapped in an object.
 */
@Test
public void testDeserializeArray() throws Exception {
    final String data = "[ 100, 500, false, 200, 400 ]";
    final Topology t = new Topology();
    TStream<String> array = t.strings(data);

    TStream<JSONObject> json = JSONStreams.deserialize(array);
    TStream<String> jsonString = JSONStreams.serialize(json);
    
    JSONArray ja = (JSONArray) JSON.parse(data);
    JSONObject jo = new JSONObject();
    jo.put("payload", ja);

    checkJsonOutput(jo, jsonString);
}
 

@Test
public void testJsonSPL() throws Exception {
    final Topology t = new Topology();
    TStream<String> example = t.strings(JSON_EXAMPLE);

    TStream<JSONObject> json = JSONStreams.deserialize(example);
    
    SPLStream spl = JSONStreams.toSPL(json);
    assertEquals(JSONSchemas.JSON, spl.getSchema());
    
    TStream<JSONObject> jsonFromSPL = spl.toJSON();  
    assertEquals(JSONObject.class, jsonFromSPL.getTupleClass());
    assertEquals(JSONObject.class, jsonFromSPL.getTupleType());

    TStream<String> jsonString = JSONStreams.serialize(jsonFromSPL);

    checkJsonOutput(JSON.parse(JSON_EXAMPLE), jsonString);
}
 

@Override
public JSONObject apply(String tuple) {
    try {
        JSONArtifact artifact = JSON.parse(tuple);
        if (artifact instanceof JSONObject)
            return (JSONObject) artifact;
        JSONObject wrapper = new JSONObject();
        wrapper.put(PAYLOAD, artifact);
        return wrapper;
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

/**
 * Convert an example JSON as a String back to a String through JSON
 * deserialization and serialization.
 */
@Test
public void testSimpleJson() throws Exception {
    final Topology t = new Topology("SimpleJson");
    TStream<String> example = t.strings(JSON_EXAMPLE);

    TStream<JSONObject> json = JSONStreams.deserialize(example);
    
    assertEquals(JSONObject.class, json.getTupleClass());
    assertEquals(JSONObject.class, json.getTupleType());
    
    
    TStream<String> jsonString = JSONStreams.serialize(json);
    
    assertEquals(String.class, jsonString.getTupleClass());
    assertEquals(String.class, jsonString.getTupleType());


    checkJsonOutput(JSON.parse(JSON_EXAMPLE), jsonString);
}