@@ -22,34 +22,77 @@ import org.apache.accumulo.core.data.{Value => AValue}
2222import org .apache .avro .io ._
2323import org .geotools .data .DataUtilities
2424import org .locationtech .geomesa .core .data .FeatureEncoding .FeatureEncoding
25- import org .locationtech .geomesa .feature .{AvroSimpleFeatureWriter , FeatureSpecificReader }
25+ import org .locationtech .geomesa .feature .{AvroSimpleFeatureFactory , AvroSimpleFeature , AvroSimpleFeatureWriter , FeatureSpecificReader }
2626import org .locationtech .geomesa .utils .text .ObjectPoolFactory
2727import org .opengis .feature .simple .{SimpleFeature , SimpleFeatureType }
2828
2929
30+ trait HasEncoding {
31+ def encoding : FeatureEncoding
32+ }
33+
3034/**
31- * Responsible for collecting data-entries that share an identifier, and
32- * when done, collapsing these rows into a single SimpleFeature.
33- *
34- * All encoding/decoding/serializing of features should be done through this
35- * single class to allow future versions of serialization instead of scattering
36- * knowledge of how the serialization is done through geomesa codebase.
35+ * Interface to encode SimpleFeatures with a configurable serialization format.
3736 *
3837 * A SimpleFeatureEncoder is bound to a given SimpleFeatureType since serialization
3938 * may depend upon the schema of the feature type.
4039 *
4140 * SimpleFeatureEncoder classes may not be thread safe and should generally be used
42- * as instance variables for performance reasons. They can serialize and deserialize
43- * multiple features.
41+ * as instance variables for performance reasons.
4442 */
45- trait SimpleFeatureEncoder {
43+ trait SimpleFeatureEncoder extends HasEncoding {
4644 def encode (feature : SimpleFeature ): Array [Byte ]
45+ }
46+
47+ /**
48+ * Interface to read SimpleFeatures with a configurable serialization format.
49+ *
50+ * A SimpleFeatureDecoder is bound to a given SimpleFeatureType since serialization
51+ * may depend upon the schema of the feature type.
52+ *
53+ * SimpleFeatureDecoder classes may not be thread safe and should generally be used
54+ * as instance variables for performance reasons.
55+ */
56+ trait SimpleFeatureDecoder extends HasEncoding {
4757 def decode (featureValue : AValue ): SimpleFeature = decode(featureValue.get)
4858 def decode (featureBytes : Array [Byte ]): SimpleFeature
4959 def extractFeatureId (value : AValue ): String = extractFeatureId(value.get)
5060 def extractFeatureId (bytes : Array [Byte ]): String
51- def getName = getEncoding.toString
52- def getEncoding : FeatureEncoding
61+ }
62+
63+ object SimpleFeatureDecoder {
64+ def apply (sft : SimpleFeatureType , encoding : FeatureEncoding ) =
65+ encoding match {
66+ case FeatureEncoding .AVRO => new AvroFeatureDecoder (sft)
67+ case FeatureEncoding .TEXT => new TextFeatureDecoder (sft)
68+ }
69+
70+ def apply (originalSft : SimpleFeatureType ,
71+ projectedSft : SimpleFeatureType ,
72+ encoding : FeatureEncoding ) =
73+ encoding match {
74+ case FeatureEncoding .AVRO => new ProjectingAvroFeatureDecoder (originalSft, projectedSft)
75+ case FeatureEncoding .TEXT => new ProjectingTextDecoder (originalSft, projectedSft)
76+ }
77+
78+ def apply (sft : SimpleFeatureType , encoding : String ): SimpleFeatureDecoder =
79+ SimpleFeatureDecoder (sft, FeatureEncoding .withName(encoding))
80+
81+ def apply (originalSft : SimpleFeatureType ,
82+ projectedSft : SimpleFeatureType ,
83+ encoding : String ): SimpleFeatureDecoder =
84+ SimpleFeatureDecoder (originalSft, projectedSft, FeatureEncoding .withName(encoding))
85+ }
86+
87+ object SimpleFeatureEncoder {
88+ def apply (sft : SimpleFeatureType , encoding : FeatureEncoding ) =
89+ encoding match {
90+ case FeatureEncoding .AVRO => new AvroFeatureEncoder (sft)
91+ case FeatureEncoding .TEXT => new TextFeatureEncoder (sft)
92+ }
93+
94+ def apply (sft : SimpleFeatureType , encoding : String ): SimpleFeatureEncoder =
95+ SimpleFeatureEncoder (sft, FeatureEncoding .withName(encoding))
5396}
5497
5598object FeatureEncoding extends Enumeration {
@@ -58,18 +101,38 @@ object FeatureEncoding extends Enumeration {
58101 val TEXT = Value (" text"
59102}
60103
61- class TextFeatureEncoder (sft : SimpleFeatureType ) extends SimpleFeatureEncoder {
62- override def encode (feature: SimpleFeature ): Array [Byte ] = ThreadSafeDataUtilities .encodeFeature(feature).getBytes
104+ class TextFeatureEncoder (sft : SimpleFeatureType ) extends SimpleFeatureEncoder {
105+ override def encode (feature: SimpleFeature ): Array [Byte ] =
106+ ThreadSafeDataUtilities .encodeFeature(feature).getBytes
63107
64- override def decode (bytes : Array [Byte ]) = ThreadSafeDataUtilities .createFeature(sft, new String (bytes))
108+ override def encoding : FeatureEncoding = FeatureEncoding .TEXT
109+ }
110+
111+ class TextFeatureDecoder (sft : SimpleFeatureType ) extends SimpleFeatureDecoder {
112+ override def decode (bytes : Array [Byte ]) =
113+ ThreadSafeDataUtilities .createFeature(sft, new String (bytes))
65114
66115 // This is derived from the knowledge of the GeoTools encoding in DataUtilities
67116 override def extractFeatureId (bytes : Array [Byte ]): String = {
68117 val featureString = new String (bytes)
69118 featureString.substring(0 , featureString.indexOf(" ="
70119 }
71120
72- override def getEncoding : FeatureEncoding = FeatureEncoding .TEXT
121+ override def encoding : FeatureEncoding = FeatureEncoding .TEXT
122+ }
123+
124+ class ProjectingTextDecoder (original : SimpleFeatureType , projected : SimpleFeatureType )
125+ extends TextFeatureDecoder (original) {
126+
127+ private val fac = AvroSimpleFeatureFactory .featureBuilder(projected)
128+ private val attrs = DataUtilities .attributeNames(projected)
129+
130+ override def decode (bytes : Array [Byte ]) = {
131+ val sf = super .decode(bytes)
132+ fac.reset()
133+ attrs.foreach { attr => fac.set(attr, sf.getAttribute(attr)) }
134+ fac.buildFeature(sf.getID)
135+ }
73136}
74137
75138/**
@@ -89,44 +152,44 @@ object ThreadSafeDataUtilities {
89152 }
90153}
91154
92- /**
93- * Encode features as avro making reuse of binary decoders and encoders
94- * as well as a custom datum writer and reader
95- *
96- * This class is NOT threadsafe and cannot be shared across multiple threads.
97- *
98- * @param sft
99- */
100155class AvroFeatureEncoder (sft : SimpleFeatureType ) extends SimpleFeatureEncoder {
101156
102157 private val writer = new AvroSimpleFeatureWriter (sft)
103- private val reader = FeatureSpecificReader (sft)
104158
105159 // Encode using a direct binary encoder that is reused. No need to buffer
106160 // small simple features. Reuse a common BAOS as well.
107161 private val baos = new ByteArrayOutputStream ()
108- private var reusableEncoder : DirectBinaryEncoder = null
162+ private var reuse : DirectBinaryEncoder = null
163+
109164 override def encode (feature : SimpleFeature ): Array [Byte ] = {
110165 baos.reset()
111- reusableEncoder = EncoderFactory .get().directBinaryEncoder(baos, reusableEncoder ).asInstanceOf [DirectBinaryEncoder ]
112- writer.write(feature, reusableEncoder )
113- reusableEncoder .flush()
166+ reuse = EncoderFactory .get().directBinaryEncoder(baos, reuse ).asInstanceOf [DirectBinaryEncoder ]
167+ writer.write(feature, reuse )
168+ reuse .flush()
114169 baos.toByteArray
115170 }
116171
172+ override def encoding : FeatureEncoding = FeatureEncoding .AVRO
173+ }
174+
175+ class ProjectingAvroFeatureDecoder (original : SimpleFeatureType , projected : SimpleFeatureType )
176+ extends SimpleFeatureDecoder {
177+
178+ private val reader = new FeatureSpecificReader (original, projected)
179+
117180 override def decode (bytes : Array [Byte ]) = decode(new ByteArrayInputStream (bytes))
118181
119- // Use a direct binary encoder that is reused. No need to buffer simple features
120- // since they are small and no stream read-ahead is required
121- private var reusableDecoder : BinaryDecoder = null
182+ private var reuse : BinaryDecoder = null
183+
122184 def decode (is : InputStream ) = {
123- reusableDecoder = DecoderFactory .get().directBinaryDecoder(is, reusableDecoder )
124- reader.read(null , reusableDecoder )
185+ reuse = DecoderFactory .get().directBinaryDecoder(is, reuse )
186+ reader.read(null , reuse )
125187 }
126188
127189 override def extractFeatureId (bytes : Array [Byte ]) =
128- FeatureSpecificReader .extractId(new ByteArrayInputStream (bytes), reusableDecoder )
190+ FeatureSpecificReader .extractId(new ByteArrayInputStream (bytes), reuse )
129191
130- override def getEncoding : FeatureEncoding = FeatureEncoding .AVRO
192+ override def encoding : FeatureEncoding = FeatureEncoding .AVRO
131193}
132194
195+ class AvroFeatureDecoder (sft : SimpleFeatureType ) extends ProjectingAvroFeatureDecoder (sft, sft)
0 commit comments