@Override
public void registerClasses(final Kryo kryo) {
  // Use existing FeatureSerializer code to serialize SimpleFeature
  // classes
  final FeatureSerializer simpleFeatureSerializer = new FeatureSerializer();
  final GridCoverageWritableSerializer gcwSerializer = new GridCoverageWritableSerializer();
  final PersistableSerializer persistSerializer = new PersistableSerializer();

  PersistableFactory.getInstance().getClassIdMapping().entrySet().forEach(
      e -> kryo.register(e.getKey(), persistSerializer, e.getValue()));

  kryo.register(GeoWaveRDD.class);
  kryo.register(GeoWaveIndexedRDD.class);
  kryo.register(Geometry.class);
  kryo.register(PreparedGeometry.class);
  kryo.register(ByteArray.class);
  kryo.register(GeoWaveInputKey.class);
  kryo.register(SimpleFeatureImpl.class, simpleFeatureSerializer);
  kryo.register(GridCoverageWritable.class, gcwSerializer);
}
 

@Before
public void setUp() throws Exception {
    // Draw a "Z"
    path = new Path2D.Double();
    path.moveTo(0, 0);
    path.lineTo(3, 0);
    path.lineTo(0, 3);
    path.lineTo(3, 3);

    product = new Product("p", "t", 4, 4);
    band = product.addBand("b", "4 * (Y-0.5) + (X-0.5) + 0.1");

    dataSourceConfig = new ProfilePlotPanel.DataSourceConfig();
    SimpleFeatureTypeBuilder ftb = new SimpleFeatureTypeBuilder();
    ftb.setName("ft");
    ftb.add("lat", Double.class);
    ftb.add("lon", Double.class);
    ftb.add("data", Double.class);
    SimpleFeatureType ft = ftb.buildFeatureType();
    DefaultFeatureCollection fc = new DefaultFeatureCollection("id", ft);
    fc.add(new SimpleFeatureImpl(new Object[]{0, 0, 0.3}, ft, new FeatureIdImpl("id1"), false));
    fc.add(new SimpleFeatureImpl(new Object[]{0, 0, 0.5}, ft, new FeatureIdImpl("id2"), false));
    fc.add(new SimpleFeatureImpl(new Object[]{0, 0, 0.7}, ft, new FeatureIdImpl("id3"), false));
    fc.add(new SimpleFeatureImpl(new Object[]{0, 0, 0.1}, ft, new FeatureIdImpl("id4"), false));
    dataSourceConfig.pointDataSource = new VectorDataNode("vd", fc);
    dataSourceConfig.dataField = ft.getDescriptor("data");
    dataSourceConfig.boxSize = 1;
    dataSourceConfig.computeInBetweenPoints = true;
}
 

@Test
public void test() throws SchemaException {
  final Kryo kryo = new Kryo();

  kryo.register(SimpleFeatureImpl.class, new FeatureSerializer());

  final SimpleFeatureType schema =
      DataUtilities.createType("testGeo", "location:Point:srid=4326,name:String");
  final List<AttributeDescriptor> descriptors = schema.getAttributeDescriptors();
  final Object[] defaults = new Object[descriptors.size()];
  int p = 0;
  for (final AttributeDescriptor descriptor : descriptors) {
    defaults[p++] = descriptor.getDefaultValue();
  }

  final SimpleFeature feature =
      SimpleFeatureBuilder.build(schema, defaults, UUID.randomUUID().toString());
  final GeometryFactory geoFactory = new GeometryFactory();

  feature.setAttribute("location", geoFactory.createPoint(new Coordinate(-45, 45)));
  final Output output = new OutputChunked();
  kryo.getSerializer(SimpleFeatureImpl.class).write(kryo, output, feature);
  final Input input = new InputChunked();
  input.setBuffer(output.getBuffer());
  final SimpleFeature f2 =
      (SimpleFeature) kryo.getSerializer(SimpleFeatureImpl.class).read(
          kryo,
          input,
          SimpleFeatureImpl.class);
  assertEquals(feature, f2);
}
 

@Override
public boolean accept(final Class<?> c) {
  return SimpleFeatureImpl.class.isAssignableFrom(c);
}
 

@Override
public Deserializer<SimpleFeatureImpl> getDeserializer(final Class<SimpleFeatureImpl> arg0) {
  return new SFDeserializer();
}
 

@Override
public Serializer<SimpleFeatureImpl> getSerializer(final Class<SimpleFeatureImpl> arg0) {
  return new SFSerializer();
}
 

@Override
public SimpleFeatureImpl deserialize(final SimpleFeatureImpl t) throws IOException {
  final FeatureWritable fw = new FeatureWritable();
  fw.readFields(dataInput);
  return (SimpleFeatureImpl) fw.getFeature();
}
 

@Override
public void serialize(final SimpleFeatureImpl t) throws IOException {
  final FeatureWritable fw = new FeatureWritable(t.getFeatureType(), t);

  fw.write(dataOutput);
}
 

/**
   * 
   * Handling non-spatial attributes only
   **/
  public void handleNonGeometricAttributes(SimpleFeatureImpl feature) throws UnsupportedEncodingException, FileNotFoundException {

	    try 
	    {
	        String featureAttribute = "featureWithoutID";
	        String featureName = "featureWithoutName";
	        String featureClass = "featureWithoutClass";

	        //Feature id
	        if (feature.getAttribute(configuration.attribute) != null) {
	          featureAttribute = feature.getAttribute(configuration.attribute).toString();
	        }

	        //Feature name
	        if (feature.getAttribute(configuration.featname) != null) {
	          featureName = feature.getAttribute(configuration.featname).toString();
	        }
	        //Feature classification
	        if (feature.getAttribute(configuration.featclass) != null) {
	          featureClass = feature.getAttribute(configuration.featclass).toString();
	        }
	   
	        if (!featureAttribute.equals(configuration.ignore)) 
	        {
	          String encodingType =
	                  URLEncoder.encode(configuration.type,
	                                    UtilsConstants.UTF_8).replace(STRING_TO_REPLACE,
	                                                                  REPLACEMENT);
	          String encodingResource =
	                  URLEncoder.encode(featureAttribute,
	                                    UtilsConstants.UTF_8).replace(STRING_TO_REPLACE,
	                                                                  REPLACEMENT);
	          String aux = encodingType + SEPARATOR + encodingResource;
	 
	          //Insert literal for name (name of feature)
	          if ((!featureName.equals(configuration.ignore)) && (!featureName.equals("")))  //NOT NULL values only
	          {
	        	  insertResourceNameLiteral(
	        			  configuration.nsUri + aux,
	        			  configuration.nsUri + Constants.NAME,
	        			  featureName, 
	        			  configuration.defaultLang );
	          }
	          
	          //Insert literal for class (type of feature)
	          if ((!featureClass.equals(configuration.ignore)) && (!featureClass.equals("")))  //NOT NULL values only
	          {
	        	  encodingResource =
	                      URLEncoder.encode(featureClass,
	                                        UtilsConstants.UTF_8).replace(STRING_TO_REPLACE,
	                                                                      REPLACEMENT);
	              //Type according to application schema
	              insertResourceTypeResource(
	                      configuration.nsUri + aux,
	                      configuration.nsUri + encodingResource);
	          
	          }
	        }
	    }
	    catch(Exception e) {}

}