IGNITE-2979: .NET: Implemented ability to use Java filter in .NET-based continuous queries.

Author: Pavel Tupitsyn

modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java

@@ -83,6 +83,7 @@
 import org.apache.ignite.internal.processors.igfs.meta.IgfsMetaFileUnlockProcessor;
 import org.apache.ignite.internal.processors.igfs.meta.IgfsMetaUpdatePropertiesProcessor;
 import org.apache.ignite.internal.processors.igfs.meta.IgfsMetaUpdateTimesProcessor;
+import org.apache.ignite.internal.processors.platform.PlatformJavaObjectFactoryProxy;
 import org.apache.ignite.internal.util.IgniteUtils;
 import org.apache.ignite.internal.util.lang.GridMapEntry;
 import org.apache.ignite.internal.util.typedef.F;
@@ -272,6 +273,8 @@ public BinaryContext(BinaryMetadataHandler metaHnd, IgniteConfiguration igniteCf
         registerPredefinedType(GridMapEntry.class, 60);
         registerPredefinedType(IgniteBiTuple.class, 61);
         registerPredefinedType(T2.class, 62);
+        registerPredefinedType(PlatformJavaObjectFactoryProxy.class,
+            GridBinaryMarshaller.PLATFORM_JAVA_OBJECT_FACTORY_PROXY);
 
         registerPredefinedType(BinaryObjectImpl.class, 0);
         registerPredefinedType(BinaryObjectOffheapImpl.class, 0);

modules/core/src/main/java/org/apache/ignite/internal/binary/GridBinaryMarshaller.java

@@ -180,6 +180,9 @@ public class GridBinaryMarshaller {
     /** */
     public static final byte LINKED_HASH_MAP = 2;
 
+    /** */
+    public static final byte PLATFORM_JAVA_OBJECT_FACTORY_PROXY = 99;
+
     /** */
     public static final int OBJECT_TYPE_ID = -1;
 

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformDefaultJavaObjectFactory.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.platform;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.Map;
+
+/**
+ * Default Java object factory implementation.
+ */
+public class PlatformDefaultJavaObjectFactory<T> implements PlatformJavaObjectFactoryEx<T> {
+    /** Class name. */
+    private String clsName;
+
+    /** Properties. */
+    private Map<String, Object> props;
+
+    /** {@inheritDoc} */
+    @Override public void initialize(@Nullable Object payload, @Nullable Map<String, Object> props) {
+        if (payload == null)
+            throw new IgniteException("Java object class name is not provided.");
+
+        assert payload instanceof String;
+
+        clsName = (String)payload;
+
+        this.props = props;
+    }
+
+    /** {@inheritDoc} */
+    @Override public T create() {
+        T res = PlatformUtils.createJavaObject(clsName);
+
+        PlatformUtils.initializeJavaObject(res, clsName, props, null);
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(PlatformDefaultJavaObjectFactory.class, this);
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformJavaObjectFactoryEx.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.platform;
+
+import org.apache.ignite.platform.PlatformJavaObjectFactory;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.Map;
+
+/**
+ * Extended Java object factory interface to handle special cases.
+ */
+public interface PlatformJavaObjectFactoryEx<T> extends PlatformJavaObjectFactory<T> {
+    /**
+     * Initialize factory.
+     *
+     * @param payload Optional payload.
+     * @param props Optional properties.
+     */
+    public void initialize(@Nullable Object payload, @Nullable Map<String, Object> props);
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformJavaObjectFactoryProxy.java

@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.platform;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.binary.BinaryReader;
+import org.apache.ignite.binary.BinaryWriter;
+import org.apache.ignite.binary.Binarylizable;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.binary.BinaryRawReaderEx;
+import org.apache.ignite.internal.binary.BinaryRawWriterEx;
+import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.platform.PlatformJavaObjectFactory;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Wrapper for Java object factory.
+ */
+public class PlatformJavaObjectFactoryProxy implements Externalizable, Binarylizable {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** User-defined type. */
+    public static final int TYP_USER = 0;
+
+    /** Default factory. */
+    public static final int TYP_DEFAULT = 1;
+
+    /** Factory type. */
+    private int factoryTyp;
+
+    /** Class name. */
+    private String clsName;
+
+    /** Optional payload for special factory types. */
+    @GridToStringExclude
+    private Object payload;
+
+    /** Properties. */
+    @GridToStringExclude
+    private Map<String, Object> props;
+
+    /**
+     * Default constructor.
+     */
+    public PlatformJavaObjectFactoryProxy() {
+        // No-op.
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param factoryTyp Factory type.
+     * @param clsName Class name.
+     * @param payload Payload.
+     * @param props Properties.
+     */
+    public PlatformJavaObjectFactoryProxy(int factoryTyp, @Nullable String clsName, @Nullable Object payload,
+        @Nullable Map<String, Object> props) {
+        this.factoryTyp = factoryTyp;
+        this.clsName = clsName;
+        this.payload = payload;
+        this.props = props;
+    }
+
+    /**
+     * Get factory instance.
+     *
+     * @param ctx Kernal context for injections.
+     * @return Factory instance.
+     */
+    public PlatformJavaObjectFactory factory(GridKernalContext ctx) {
+        // Create factory.
+        Object res;
+
+        switch (factoryTyp) {
+            case TYP_DEFAULT:
+                res = new PlatformDefaultJavaObjectFactory();
+
+                break;
+
+            case TYP_USER:
+                res = PlatformUtils.createJavaObject(clsName);
+
+                break;
+
+            default:
+                throw new IgniteException("Unsupported Java object factory type: " + factoryTyp);
+        }
+
+        // Initialize factory.
+        if (res instanceof PlatformJavaObjectFactoryEx)
+            ((PlatformJavaObjectFactoryEx)res).initialize(payload, props);
+        else {
+            PlatformUtils.initializeJavaObject(res, clsName, props, ctx);
+
+            if (!(res instanceof PlatformJavaObjectFactory))
+                res = new PlatformJavaObjectSingletonFactory<>(res);
+        }
+
+        return (PlatformJavaObjectFactory)res;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException {
+        BinaryRawWriterEx rawWriter = (BinaryRawWriterEx)writer.rawWriter();
+
+        rawWriter.writeInt(factoryTyp);
+        rawWriter.writeString(clsName);
+        rawWriter.writeObjectDetached(payload);
+
+        if (props != null) {
+            rawWriter.writeInt(props.size());
+
+            for (Map.Entry<String, Object> prop : props.entrySet()) {
+                rawWriter.writeString(prop.getKey());
+                rawWriter.writeObjectDetached(prop.getValue());
+            }
+        }
+        else
+            rawWriter.writeInt(0);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readBinary(BinaryReader reader) throws BinaryObjectException {
+        BinaryRawReaderEx rawReader = (BinaryRawReaderEx)reader.rawReader();
+
+        factoryTyp = rawReader.readInt();
+        clsName = rawReader.readString();
+        payload = rawReader.readObjectDetached();
+
+        int propsSize = rawReader.readInt();
+
+        if (propsSize > 0) {
+            props = new HashMap<>(propsSize);
+
+            for (int i = 0; i < propsSize; i++) {
+                String key = rawReader.readString();
+                Object val = rawReader.readObjectDetached();
+
+                props.put(key, val);
+            }
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeInt(factoryTyp);
+        U.writeString(out, clsName);
+        out.writeObject(payload);
+        U.writeMap(out, props);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+        factoryTyp = in.readInt();
+        clsName = U.readString(in);
+        payload = in.readObject();
+        props = U.readMap(in);
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(PlatformJavaObjectFactoryProxy.class, this);
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformJavaObjectSingletonFactory.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.platform;
+
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.platform.PlatformJavaObjectFactory;
+
+/**
+ * Singleton factory.
+ */
+public class PlatformJavaObjectSingletonFactory<T> implements PlatformJavaObjectFactory<T> {
+    /** Instance. */
+    private final T instance;
+
+    /**
+     * Constructor.
+     *
+     * @param instance Instance.
+     */
+    public PlatformJavaObjectSingletonFactory(T instance) {
+        this.instance = instance;
+    }
+
+    /** {@inheritDoc} */
+    @Override public T create() {
+        return instance;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(PlatformJavaObjectSingletonFactory.class, this);
+    }
+}