Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2202#discussion_r70449634
--- Diff:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java ---
@@ -454,4 +458,126 @@ static String generateStateName(final String name,
final int index) {
return name + "_" + index;
}
}
+
+ /**
+ * {@link TypeSerializer} for {@link NFA} that uses Java Serialization.
+ */
+ public static class Serializer<T> extends TypeSerializer<NFA<T>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<NFA<T>> duplicate() {
+ return this;
+ }
+
+ @Override
+ public NFA<T> createInstance() {
+ return null;
+ }
+
+ @Override
+ public NFA<T> copy(NFA<T> from) {
+ try {
+ ByteArrayOutputStream baos = new
ByteArrayOutputStream();
+ ObjectOutputStream oos = new
ObjectOutputStream(baos);
+
+ oos.writeObject(from);
+
+ oos.close();
+ baos.close();
+
+ byte[] data = baos.toByteArray();
+
+ ByteArrayInputStream bais = new
ByteArrayInputStream(data);
+ ObjectInputStream ois = new
ObjectInputStream(bais);
+
+ @SuppressWarnings("unchecked")
+ NFA<T> copy = (NFA<T>) ois.readObject();
+ return copy;
+ } catch (IOException|ClassNotFoundException e) {
+ throw new RuntimeException("Could not copy
NFA.", e);
+ }
+ }
+
+ @Override
+ public NFA<T> copy(NFA<T> from, NFA<T> reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return 0;
+ }
+
+ @Override
+ public void serialize(NFA<T> record, DataOutputView target)
throws IOException {
+ try {
+ ByteArrayOutputStream baos = new
ByteArrayOutputStream();
+ ObjectOutputStream oos = new
ObjectOutputStream(baos);
+
+ oos.writeObject(record);
+
+ oos.close();
+ baos.close();
+
+ byte[] data = baos.toByteArray();
+
+ target.writeInt(data.length);
+ target.write(data);
--- End diff --
Can't we use the `DataOutputViewStream`?
```
ObjectOutputStream oos = new ObjectOutputStream(new
DataOutputViewStream(target));
oos.writeObject(record);
oos.close();
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
