diff --git a/src/config/connect-knet-sink.properties b/src/config/connect-knet-sink.properties new file mode 100644 index 0000000000..22ff9b967f --- /dev/null +++ b/src/config/connect-knet-sink.properties @@ -0,0 +1,20 @@ +# 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. + +name=local-knet-sink +connector.class=KNetSinkConnector +tasks.max=1 +knet.dotnet.classname=MASES.KNetTemplate.KNetConnect.KNetConnectSink, knetConnectSink +topics=topic-perf \ No newline at end of file diff --git a/src/config/connect-knet-source.properties b/src/config/connect-knet-source.properties new file mode 100644 index 0000000000..47cc28b0fe --- /dev/null +++ b/src/config/connect-knet-source.properties @@ -0,0 +1,20 @@ +# 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. + +name=local-knet-source +connector.class=KNetSourceConnector +tasks.max=1 +knet.dotnet.classname=MASES.KNetTemplate.KNetConnect.KNetConnectSource, knetConnectSource +topic=topic-perf \ No newline at end of file diff --git a/src/java/knet/pom.xml b/src/java/knet/pom.xml index 581aa7dd7a..b73df91a82 100644 --- a/src/java/knet/pom.xml +++ b/src/java/knet/pom.xml @@ -9,7 +9,7 @@ mases.knet Interface bridging implementation for Apache Kafka https://github.com/masesgroup/KNet - 1.2.4.0 + 1.3.0.0 @@ -40,7 +40,7 @@ 8 8 ${basedir}/classpathfile.classpath - 3.1.0 + 3.1.1 @@ -228,6 +228,11 @@ 4.13.1 test + + org.slf4j + slf4j-api + 1.7.36 + org.apache.kafka kafka-clients diff --git a/src/java/knet/src/main/java/org/mases/knet/connect/KNetConnectProxy.java b/src/java/knet/src/main/java/org/mases/knet/connect/KNetConnectProxy.java new file mode 100644 index 0000000000..775a8cd20e --- /dev/null +++ b/src/java/knet/src/main/java/org/mases/knet/connect/KNetConnectProxy.java @@ -0,0 +1,89 @@ +/* + * Copyright 2022 MASES s.r.l. + * + * Licensed 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. + * + * Refer to LICENSE for more information. + */ + +package org.mases.knet.connect; + +import org.apache.kafka.common.config.AbstractConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigException; +import org.mases.jcobridge.*; +import org.mases.knet.connect.source.KNetSourceTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Map; + +public class KNetConnectProxy { + private static final Logger log = LoggerFactory.getLogger(KNetSourceTask.class); + + public static final String DOTNET_CLASSNAME_CONFIG = "knet.dotnet.classname"; + + public static final ConfigDef CONFIG_DEF = new ConfigDef() + .define(DOTNET_CLASSNAME_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH, ".NET class name in the form usable from .NET like \"classname, assembly name\"."); + + static JCObject knetConnectProxy = null; + static String sinkConnectorName = null; + static JCObject sinkConnector = null; + static String sourceConnectorName = null; + static JCObject sourceConnector = null; + + static synchronized JCObject getConnectProxy() throws JCException, IOException { + if (knetConnectProxy == null) { + JCOBridge.Initialize(); + knetConnectProxy = (JCObject) JCOBridge.GetCLRGlobal("KNetConnectProxy"); + } + return knetConnectProxy; + } + + public static synchronized boolean initializeSinkConnector(Map props) throws JCException, IOException { + AbstractConfig parsedConfig = new AbstractConfig(CONFIG_DEF, props); + String className = parsedConfig.getString(DOTNET_CLASSNAME_CONFIG); + if (className == null) + throw new ConfigException("'classname' in KNetSinkConnector configuration requires a definition"); + return (boolean) getConnectProxy().Invoke("AllocateSinkConnector", className); + } + + public static synchronized boolean initializeSourceConnector(Map props) throws JCException, IOException { + AbstractConfig parsedConfig = new AbstractConfig(CONFIG_DEF, props); + String className = parsedConfig.getString(DOTNET_CLASSNAME_CONFIG); + if (className == null) + throw new ConfigException("'classname' in KNetSinkConnector configuration requires a definition"); + return (boolean) getConnectProxy().Invoke("AllocateSourceConnector", className); + } + + public static synchronized JCObject getSinkConnector() throws JCException, IOException { + if (sinkConnector == null) { + sinkConnectorName = (String) getConnectProxy().Invoke("SinkConnectorName"); + if (sinkConnectorName != null) { + sinkConnector = (JCObject) JCOBridge.GetCLRGlobal(sinkConnectorName); + } + } + return sinkConnector; + } + + public static synchronized JCObject getSourceConnector() throws JCException, IOException { + if (sourceConnector == null) { + sourceConnectorName = (String) getConnectProxy().Invoke("SourceConnectorName"); + if (sourceConnectorName != null) { + sourceConnector = (JCObject) JCOBridge.GetCLRGlobal(sourceConnectorName); + } + } + return sourceConnector; + } +} diff --git a/src/java/knet/src/main/java/org/mases/knet/connect/sink/KNetSinkConnector.java b/src/java/knet/src/main/java/org/mases/knet/connect/sink/KNetSinkConnector.java new file mode 100644 index 0000000000..90666a44ee --- /dev/null +++ b/src/java/knet/src/main/java/org/mases/knet/connect/sink/KNetSinkConnector.java @@ -0,0 +1,130 @@ +/* + * Copyright 2022 MASES s.r.l. + * + * Licensed 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. + * + * Refer to LICENSE for more information. + */ + +package org.mases.knet.connect.sink; + +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.connect.connector.Task; +import org.apache.kafka.connect.sink.SinkConnector; +import org.mases.jcobridge.*; +import org.mases.knet.connect.KNetConnectProxy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class KNetSinkConnector extends SinkConnector { + private static final Logger log = LoggerFactory.getLogger(KNetSinkConnector.class); + + private static final String registrationName = "KNetSinkConnector"; + + Object dataToExchange = null; + + public Object getDataToExchange() { + return dataToExchange; + } + + public void setDataToExchange(Object dte) { + dataToExchange = dte; + } + + @Override + public void start(Map props) { + try { + if (!KNetConnectProxy.initializeSinkConnector(props)) { + log.error("Failed Invoke of \"initializeSinkConnector\""); + throw new ConfigException("Failed Invoke of \"initializeSinkConnector\""); + } else { + JCOBridge.RegisterJVMGlobal(registrationName, this); + try { + dataToExchange = props; + JCObject sink = KNetConnectProxy.getSinkConnector(); + if (sink == null) throw new ConfigException("getSinkConnector returned null."); + sink.Invoke("StartInternal"); + } finally { + dataToExchange = null; + } + } + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"start\"", jcne); + } + } + + @Override + public Class taskClass() { + return KNetSinkTask.class; + } + + @Override + public List> taskConfigs(int maxTasks) { + ArrayList> configs = new ArrayList<>(); + for (int i = 0; i < maxTasks; i++) { + Map config = new HashMap<>(); + try { + dataToExchange = config; + JCObject sink = KNetConnectProxy.getSinkConnector(); + if (sink == null) throw new ConfigException("getSinkConnector returned null."); + sink.Invoke("TaskConfigsInternal", i); + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"start\"", jcne); + } finally { + dataToExchange = null; + } + configs.add(config); + } + return configs; + } + + @Override + public void stop() { + try { + try { + JCObject sink = KNetConnectProxy.getSinkConnector(); + if (sink == null) throw new ConfigException("getSinkConnector returned null."); + sink.Invoke("StopInternal"); + } finally { + JCOBridge.UnregisterJVMGlobal(registrationName); + } + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"stop\"", jcne); + } + } + + @Override + public ConfigDef config() { + return KNetConnectProxy.CONFIG_DEF; + } + + @Override + public String version() { + try { + JCObject sink = KNetConnectProxy.getSinkConnector(); + if (sink != null) { + return (String) sink.Invoke("VersionInternal"); + } + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"version\"", jcne); + } + return "NOT AVAILABLE"; + } +} diff --git a/src/java/knet/src/main/java/org/mases/knet/connect/sink/KNetSinkTask.java b/src/java/knet/src/main/java/org/mases/knet/connect/sink/KNetSinkTask.java new file mode 100644 index 0000000000..7757237b8f --- /dev/null +++ b/src/java/knet/src/main/java/org/mases/knet/connect/sink/KNetSinkTask.java @@ -0,0 +1,108 @@ +/* + * Copyright 2022 MASES s.r.l. + * + * Licensed 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. + * + * Refer to LICENSE for more information. + */ + +package org.mases.knet.connect.sink; + +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.connect.sink.SinkRecord; +import org.apache.kafka.connect.sink.SinkTask; +import org.apache.kafka.connect.sink.SinkTaskContext; +import org.mases.jcobridge.*; +import org.mases.knet.connect.KNetConnectProxy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +public class KNetSinkTask extends SinkTask { + private static final Logger log = LoggerFactory.getLogger(KNetSinkTask.class); + + static final AtomicLong taskId = new AtomicLong(0); + JCObject sinkTask = null; + + Object dataToExchange = null; + + public Object getDataToExchange() { + return dataToExchange; + } + + public void setDataToExchange(Object dte) { + dataToExchange = dte; + } + + public KNetSinkTask() throws ConfigException, JCException, IOException { + super(); + long taskid = taskId.incrementAndGet(); + JCOBridge.RegisterJVMGlobal(String.format("KNetSinkTask_%d", taskid), this); + JCObject sink = KNetConnectProxy.getSinkConnector(); + if (sink == null) throw new ConfigException("getSinkConnector returned null."); + sinkTask = (JCObject) sink.Invoke("AllocateTask", taskid); + } + + @Override + public String version() { + try { + if (sinkTask != null) { + return (String) sinkTask.Invoke("VersionInternal"); + } + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"version\"", jcne); + } + return "NOT AVAILABLE"; + } + + @Override + public void start(Map map) { + try { + try { + dataToExchange = map; + sinkTask.Invoke("StartInternal"); + } finally { + dataToExchange = null; + } + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"start\"", jcne); + } + } + + @Override + public void put(Collection collection) { + try { + try { + dataToExchange = collection; + sinkTask.Invoke("PutInternal"); + } finally { + dataToExchange = null; + } + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"put\"", jcne); + } + } + + @Override + public void stop() { + try { + sinkTask.Invoke("StopInternal"); + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"stop\"", jcne); + } + } +} diff --git a/src/java/knet/src/main/java/org/mases/knet/connect/source/KNetSourceConnector.java b/src/java/knet/src/main/java/org/mases/knet/connect/source/KNetSourceConnector.java new file mode 100644 index 0000000000..669c033c6f --- /dev/null +++ b/src/java/knet/src/main/java/org/mases/knet/connect/source/KNetSourceConnector.java @@ -0,0 +1,130 @@ +/* + * Copyright 2022 MASES s.r.l. + * + * Licensed 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. + * + * Refer to LICENSE for more information. + */ + +package org.mases.knet.connect.source; + +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.connect.connector.Task; +import org.apache.kafka.connect.source.SourceConnector; +import org.mases.jcobridge.*; +import org.mases.knet.connect.KNetConnectProxy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class KNetSourceConnector extends SourceConnector { + private static final Logger log = LoggerFactory.getLogger(KNetSourceConnector.class); + + private static final String registrationName = "KNetSourceConnector"; + + Object dataToExchange = null; + + public Object getDataToExchange() { + return dataToExchange; + } + + public void setDataToExchange(Object dte) { + dataToExchange = dte; + } + + @Override + public void start(Map props) { + try { + if (!KNetConnectProxy.initializeSourceConnector(props)) { + log.error("Failed Invoke of \"initializeSourceConnector\""); + throw new ConfigException("Failed Invoke of \"initializeSourceConnector\""); + } else { + JCOBridge.RegisterJVMGlobal(registrationName, this); + try { + dataToExchange = props; + JCObject source = KNetConnectProxy.getSourceConnector(); + if (source == null) throw new ConfigException("getSourceConnector returned null."); + source.Invoke("StartInternal"); + } finally { + dataToExchange = null; + } + } + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"start\"", jcne); + } + } + + @Override + public Class taskClass() { + return KNetSourceTask.class; + } + + @Override + public List> taskConfigs(int maxTasks) { + ArrayList> configs = new ArrayList<>(); + for (int i = 0; i < maxTasks; i++) { + Map config = new HashMap<>(); + try { + dataToExchange = config; + JCObject source = KNetConnectProxy.getSourceConnector(); + if (source == null) throw new ConfigException("getSourceConnector returned null."); + source.Invoke("TaskConfigsInternal", i); + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"start\"", jcne); + } finally { + dataToExchange = null; + } + configs.add(config); + } + return configs; + } + + @Override + public void stop() { + try { + try { + JCObject source = KNetConnectProxy.getSourceConnector(); + if (source == null) throw new ConfigException("getSourceConnector returned null."); + source.Invoke("StopInternal"); + } finally { + JCOBridge.UnregisterJVMGlobal(registrationName); + } + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"stop\"", jcne); + } + } + + @Override + public ConfigDef config() { + return KNetConnectProxy.CONFIG_DEF; + } + + @Override + public String version() { + try { + JCObject source = KNetConnectProxy.getSourceConnector(); + if (source != null) { + return (String) source.Invoke("VersionInternal"); + } + } catch (JCException | IOException jcne) { + log.error("Failed Invoke of \"version\"", jcne); + } + return "NOT AVAILABLE"; + } +} diff --git a/src/java/knet/src/main/java/org/mases/knet/connect/source/KNetSourceTask.java b/src/java/knet/src/main/java/org/mases/knet/connect/source/KNetSourceTask.java new file mode 100644 index 0000000000..b57bfd60f4 --- /dev/null +++ b/src/java/knet/src/main/java/org/mases/knet/connect/source/KNetSourceTask.java @@ -0,0 +1,104 @@ +/* + * Copyright 2022 MASES s.r.l. + * + * Licensed 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. + * + * Refer to LICENSE for more information. + */ + +package org.mases.knet.connect.source; + +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.connect.source.SourceRecord; +import org.apache.kafka.connect.source.SourceTask; +import org.mases.jcobridge.*; +import org.mases.knet.connect.KNetConnectProxy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +public class KNetSourceTask extends SourceTask { + private static final Logger log = LoggerFactory.getLogger(KNetSourceTask.class); + static final AtomicLong taskId = new AtomicLong(0); + + JCObject sourceTask = null; + + Object dataToExchange = null; + + public Object getDataToExchange() { + return dataToExchange; + } + + public void setDataToExchange(Object dte) { + dataToExchange = dte; + } + + public KNetSourceTask() throws ConfigException, JCException, IOException { + super(); + long taskid = taskId.incrementAndGet(); + JCOBridge.RegisterJVMGlobal(String.format("KNetSourceTask_%d", taskid), this); + JCObject source = KNetConnectProxy.getSourceConnector(); + if (source == null) throw new ConfigException("getSourceConnector returned null."); + sourceTask = (JCObject) source.Invoke("AllocateTask", taskid); + } + + @Override + public String version() { + try { + if (sourceTask != null) { + return (String) sourceTask.Invoke("VersionInternal"); + } + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"version\"", jcne); + } + return "NOT AVAILABLE"; + } + + @Override + public void start(Map map) { + try { + try { + dataToExchange = map; + sourceTask.Invoke("StartInternal"); + } finally { + dataToExchange = null; + } + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"start\"", jcne); + } + } + + @Override + public List poll() throws InterruptedException { + try { + Object result = sourceTask.Invoke("PollInternal"); + if (result != null) return (List) result; + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"poll\"", jcne); + } + return null; + } + + @Override + public void stop() { + try { + sourceTask.Invoke("StopInternal"); + } catch (JCNativeException jcne) { + log.error("Failed Invoke of \"stop\"", jcne); + } + } +} diff --git a/src/java/knet/src/test/java/org/mases/knet/connect/KNetConnectProxyTest.java b/src/java/knet/src/test/java/org/mases/knet/connect/KNetConnectProxyTest.java new file mode 100644 index 0000000000..9f8133f8aa --- /dev/null +++ b/src/java/knet/src/test/java/org/mases/knet/connect/KNetConnectProxyTest.java @@ -0,0 +1,18 @@ +package org.mases.knet.connect; + +import org.junit.Test; + +import static org.junit.Assert.assertTrue; + +/** + * Unit test for simple App. + */ +public class KNetConnectProxyTest { + /** + * Rigorous Test :-) + */ + @Test + public void shouldAnswerWithTrue() { + assertTrue(true); + } +} \ No newline at end of file diff --git a/src/java/knet/src/test/java/org/mases/knet/connect/sink/KNetSinkConnectorTest.java b/src/java/knet/src/test/java/org/mases/knet/connect/sink/KNetSinkConnectorTest.java new file mode 100644 index 0000000000..39cb0bd702 --- /dev/null +++ b/src/java/knet/src/test/java/org/mases/knet/connect/sink/KNetSinkConnectorTest.java @@ -0,0 +1,18 @@ +package org.mases.knet.connect.sink; + +import org.junit.Test; + +import static org.junit.Assert.assertTrue; + +/** + * Unit test for simple App. + */ +public class KNetSinkConnectorTest { + /** + * Rigorous Test :-) + */ + @Test + public void shouldAnswerWithTrue() { + assertTrue(true); + } +} \ No newline at end of file diff --git a/src/java/knet/src/test/java/org/mases/knet/connect/sink/KNetSinkTaskTest.java b/src/java/knet/src/test/java/org/mases/knet/connect/sink/KNetSinkTaskTest.java new file mode 100644 index 0000000000..a46129a093 --- /dev/null +++ b/src/java/knet/src/test/java/org/mases/knet/connect/sink/KNetSinkTaskTest.java @@ -0,0 +1,18 @@ +package org.mases.knet.connect.sink; + +import org.junit.Test; + +import static org.junit.Assert.assertTrue; + +/** + * Unit test for simple App. + */ +public class KNetSinkTaskTest { + /** + * Rigorous Test :-) + */ + @Test + public void shouldAnswerWithTrue() { + assertTrue(true); + } +} \ No newline at end of file diff --git a/src/java/knet/src/test/java/org/mases/knet/connect/source/KNetSourceConnectorTest.java b/src/java/knet/src/test/java/org/mases/knet/connect/source/KNetSourceConnectorTest.java new file mode 100644 index 0000000000..46e46dfda1 --- /dev/null +++ b/src/java/knet/src/test/java/org/mases/knet/connect/source/KNetSourceConnectorTest.java @@ -0,0 +1,18 @@ +package org.mases.knet.connect.source; + +import org.junit.Test; + +import static org.junit.Assert.assertTrue; + +/** + * Unit test for simple App. + */ +public class KNetSourceConnectorTest { + /** + * Rigorous Test :-) + */ + @Test + public void shouldAnswerWithTrue() { + assertTrue(true); + } +} \ No newline at end of file diff --git a/src/java/knet/src/test/java/org/mases/knet/connect/source/KNetSourceTaskTest.java b/src/java/knet/src/test/java/org/mases/knet/connect/source/KNetSourceTaskTest.java new file mode 100644 index 0000000000..652b54c19c --- /dev/null +++ b/src/java/knet/src/test/java/org/mases/knet/connect/source/KNetSourceTaskTest.java @@ -0,0 +1,18 @@ +package org.mases.knet.connect.source; + +import org.junit.Test; + +import static org.junit.Assert.assertTrue; + +/** + * Unit test for simple App. + */ +public class KNetSourceTaskTest { + /** + * Rigorous Test :-) + */ + @Test + public void shouldAnswerWithTrue() { + assertTrue(true); + } +} \ No newline at end of file diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/Cli/ConnectDistributed.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/Cli/ConnectDistributed.cs index 0fc0df4b66..bb34b026b7 100644 --- a/src/net/KNet/ClientSide/BridgedClasses/Connect/Cli/ConnectDistributed.cs +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/Cli/ConnectDistributed.cs @@ -29,6 +29,7 @@ public class ConnectDistributed : JCOBridge.C2JBridge.JVMBridgeMain taskDictionary = new(); + + IJavaObject reflectedConnector = null; + + public KNetConnector() + { + KNetCore.GlobalInstance.RegisterCLRGlobal(ReflectedConnectorName, this); + } + + protected T DataToExchange() + { + if (reflectedConnector == null) + { + reflectedConnector = KNetCore.GlobalInstance.GetJVMGlobal(ReflectedConnectorName); + } + return (reflectedConnector != null) ? reflectedConnector.Invoke("getDataToExchange") : throw new InvalidOperationException($"{ReflectedConnectorName} was not registered in global JVM"); + } + + public object AllocateTask(long taskId) + { + return taskDictionary.GetOrAdd(taskId, (id) => + { + KNetTask knetTask = Activator.CreateInstance(TaskClassType) as KNetTask; + knetTask.Initialize(this, id); + return knetTask; + }); + } + + public abstract string ReflectedConnectorName { get; } + + public abstract string ConnectorName { get; } + + public abstract Type TaskClassType { get; } + + public void Initialize(ConnectorContext ctx) => throw new NotImplementedException("Invoked in Java before any initialization."); + + public void Initialize(ConnectorContext ctx, List> taskConfigs) => throw new NotImplementedException("Invoked in Java before any initialization."); + + public void StartInternal() + { + Map props = DataToExchange>(); + Start(props); + } + + public abstract void Start(Map props); + + public void Reconfigure(Map props) { } + + public Class TaskClass() => throw new NotImplementedException("Invoked in Java before any initialization."); + + public void TaskConfigsInternal(int index) + { + Map props = DataToExchange>(); + TaskConfigs(index, props); + } + + public abstract void TaskConfigs(int index, Map config); + + public List> TaskConfigs(int maxTasks) => throw new NotImplementedException("Invoked using the other signature."); + + public void StopInternal() + { + Stop(); + } + + public abstract void Stop(); + + public object ValidateInternal(object connectorConfigsObj) + { + Map connectorConfigs = DataToExchange>(); + return Validate(connectorConfigs); + } + + public abstract Config Validate(Map connectorConfigs); + + public ConfigDef Config() => throw new NotImplementedException("Invoked in Java before any initialization."); + + public string Version() => throw new NotImplementedException("Invoked in Java before any initialization."); + } +} diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSinkConnector.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSinkConnector.cs new file mode 100644 index 0000000000..90c0eaf072 --- /dev/null +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSinkConnector.cs @@ -0,0 +1,29 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using System; + +namespace MASES.KNet.Connect +{ + public abstract class KNetSinkConnector : KNetConnector where TTask : KNetSinkTask + { + public sealed override string ReflectedConnectorName => "KNetSinkConnector"; + + public sealed override Type TaskClassType => typeof(TTask); + } +} diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSinkTask.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSinkTask.cs new file mode 100644 index 0000000000..c9b82f0024 --- /dev/null +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSinkTask.cs @@ -0,0 +1,36 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using Java.Util; +using MASES.KNet.Connect.Sink; + +namespace MASES.KNet.Connect +{ + public abstract class KNetSinkTask : KNetTask + { + public override string ReflectedTaskClassName => "KNetSinkTask"; + + public void PutInternal() + { + Collection collection = DataToExchange>(); + Put(collection); + } + + public abstract void Put(Collection collection); + } +} diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSourceConnector.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSourceConnector.cs new file mode 100644 index 0000000000..028358b882 --- /dev/null +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSourceConnector.cs @@ -0,0 +1,29 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using System; + +namespace MASES.KNet.Connect +{ + public abstract class KNetSourceConnector : KNetConnector where TTask : KNetSourceTask + { + public sealed override string ReflectedConnectorName => "KNetSourceConnector"; + + public sealed override Type TaskClassType => typeof(TTask); + } +} diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSourceTask.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSourceTask.cs new file mode 100644 index 0000000000..1149505862 --- /dev/null +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetSourceTask.cs @@ -0,0 +1,35 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using Java.Util; +using MASES.KNet.Connect.Source; + +namespace MASES.KNet.Connect +{ + public abstract class KNetSourceTask : KNetTask + { + public override string ReflectedTaskClassName => "KNetSourceTask"; + + public List PollInternal() + { + return Poll(); + } + + public abstract List Poll(); + } +} diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetTask.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetTask.cs new file mode 100644 index 0000000000..a387a287c1 --- /dev/null +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/KNetTask.cs @@ -0,0 +1,79 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using Java.Util; +using MASES.JCOBridge.C2JBridge.JVMInterop; +using MASES.KNet.Connect.Connector; +using System; + +namespace MASES.KNet.Connect +{ + public interface IKNetTask : ITask + { + IKNetConnector Connector { get; } + + long TaskId { get; } + } + + public abstract class KNetTask : IKNetTask + { + IKNetConnector connector; + long taskId; + IJavaObject reflectedTask = null; + + internal void Initialize(IKNetConnector connector, long taskId) + { + this.connector = connector; + this.taskId = taskId; + reflectedTask = KNetCore.GlobalInstance.GetJVMGlobal($"{ReflectedTaskClassName}_{taskId}"); + } + + protected T DataToExchange() + { + return (reflectedTask != null) ? reflectedTask.Invoke("getDataToExchange") : throw new InvalidOperationException($"{ReflectedTaskClassName} was not registered in global JVM"); + } + + public IKNetConnector Connector => connector; + + public long TaskId => taskId; + + public abstract string ReflectedTaskClassName { get; } + + public void StartInternal() + { + Map props = DataToExchange>(); + Start(props); + } + + public abstract void Start(Map props); + + public void StopInternal() + { + Stop(); + } + + public abstract void Stop(); + + public object VersionInternal() + { + return Version(); + } + + public abstract string Version(); + } +} diff --git a/src/net/KNet/ClientSide/BridgedClasses/Connect/Sink/SinkTaskContext.cs b/src/net/KNet/ClientSide/BridgedClasses/Connect/Sink/SinkTaskContext.cs index 507bc79a6e..4de202192b 100644 --- a/src/net/KNet/ClientSide/BridgedClasses/Connect/Sink/SinkTaskContext.cs +++ b/src/net/KNet/ClientSide/BridgedClasses/Connect/Sink/SinkTaskContext.cs @@ -25,6 +25,7 @@ namespace MASES.KNet.Connect.Sink public class SinkTaskContext : JVMBridgeBase { public override bool IsInterface => true; + public override string ClassName => "org.apache.kafka.connect.sink.SinkTaskContext"; public Map Configs => IExecute>("configs"); diff --git a/src/net/KNet/KNet.csproj b/src/net/KNet/KNet.csproj index a1b523b2fa..6aa2950c70 100644 --- a/src/net/KNet/KNet.csproj +++ b/src/net/KNet/KNet.csproj @@ -9,7 +9,7 @@ MASES s.r.l. MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 KNet true net461;netcoreapp3.1;net5.0;net6.0 diff --git a/src/net/KNet/KNetCore.cs b/src/net/KNet/KNetCore.cs index 07b63d0eaf..b03de72915 100644 --- a/src/net/KNet/KNetCore.cs +++ b/src/net/KNet/KNetCore.cs @@ -605,6 +605,10 @@ protected override IDictionary Options ReleasePath, ReleaseAdditionalPath, }); + +#if DEBUG + public override bool EnableDebug => true; +#endif } /// /// Directly usable implementation of diff --git a/src/net/KNetCLI/KNetCLI.csproj b/src/net/KNetCLI/KNetCLI.csproj index 094ca87881..50bf14fb8f 100644 --- a/src/net/KNetCLI/KNetCLI.csproj +++ b/src/net/KNetCLI/KNetCLI.csproj @@ -10,7 +10,7 @@ MASES s.r.l. MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 KNetCLI true net461;netcoreapp3.1;net5.0;net6.0 diff --git a/src/net/KNetCLI/KNetCLI.nuspec b/src/net/KNetCLI/KNetCLI.nuspec index 18d4ee1abd..b2c55cb561 100644 --- a/src/net/KNetCLI/KNetCLI.nuspec +++ b/src/net/KNetCLI/KNetCLI.nuspec @@ -2,7 +2,7 @@ MASES.KNetCLI - 1.2.4 + 1.3.0 KNetCLI - CLI interface of KNet MASES s.r.l. MASES s.r.l. diff --git a/src/net/KNetTemplates.sln b/src/net/KNetTemplates.sln index 0c6eb40a22..6f379959e3 100644 --- a/src/net/KNetTemplates.sln +++ b/src/net/KNetTemplates.sln @@ -1,7 +1,7 @@  Microsoft Visual Studio Solution File, Format Version 12.00 -# Visual Studio Version 16 -VisualStudioVersion = 16.0.31624.102 +# Visual Studio Version 17 +VisualStudioVersion = 17.1.32319.34 MinimumVisualStudioVersion = 10.0.40219.1 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "KNet", "KNet\KNet.csproj", "{3DB45E58-F79C-448E-A028-2CD17A55EFF6}" EndProject @@ -11,6 +11,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "knetConsumerApp", "template EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "knetPipeStreamApp", "templates\templates\knetPipeStreamApp\knetPipeStreamApp.csproj", "{D022E0DE-0B84-4C83-9530-C3B9D797D6C3}" EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "knetConnectSource", "templates\templates\knetConnectSource\knetConnectSource.csproj", "{2E5B6053-82B6-4109-97DE-BCD3C1976D8C}" +EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "knetConnectSink", "templates\templates\knetConnectSink\knetConnectSink.csproj", "{FC5B7EBE-067F-43EB-A530-DCA2263FCBA8}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -33,6 +37,14 @@ Global {D022E0DE-0B84-4C83-9530-C3B9D797D6C3}.Debug|Any CPU.Build.0 = Debug|Any CPU {D022E0DE-0B84-4C83-9530-C3B9D797D6C3}.Release|Any CPU.ActiveCfg = Release|Any CPU {D022E0DE-0B84-4C83-9530-C3B9D797D6C3}.Release|Any CPU.Build.0 = Release|Any CPU + {2E5B6053-82B6-4109-97DE-BCD3C1976D8C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {2E5B6053-82B6-4109-97DE-BCD3C1976D8C}.Debug|Any CPU.Build.0 = Debug|Any CPU + {2E5B6053-82B6-4109-97DE-BCD3C1976D8C}.Release|Any CPU.ActiveCfg = Release|Any CPU + {2E5B6053-82B6-4109-97DE-BCD3C1976D8C}.Release|Any CPU.Build.0 = Release|Any CPU + {FC5B7EBE-067F-43EB-A530-DCA2263FCBA8}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {FC5B7EBE-067F-43EB-A530-DCA2263FCBA8}.Debug|Any CPU.Build.0 = Debug|Any CPU + {FC5B7EBE-067F-43EB-A530-DCA2263FCBA8}.Release|Any CPU.ActiveCfg = Release|Any CPU + {FC5B7EBE-067F-43EB-A530-DCA2263FCBA8}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/src/net/templates/templates/knetConnectSink/.template.config/template.json b/src/net/templates/templates/knetConnectSink/.template.config/template.json new file mode 100644 index 0000000000..f1b8d20b19 --- /dev/null +++ b/src/net/templates/templates/knetConnectSink/.template.config/template.json @@ -0,0 +1,12 @@ +{ + "$schema": "http://json.schemastore.org/template", + "author": "MASES s.r.l.", + "classifications": [ "Common", "Library" ], + "identity": "MASES.KNetTemplate.KNetConnect", + "name": "Library templates: KNet Connect Sink project", + "shortName": "knetConnectSink", + "tags": { + "language": "C#", + "type": "project" + } +} \ No newline at end of file diff --git a/src/net/templates/templates/knetConnectSink/KNetConnectSink.cs b/src/net/templates/templates/knetConnectSink/KNetConnectSink.cs new file mode 100644 index 0000000000..de5b364c5b --- /dev/null +++ b/src/net/templates/templates/knetConnectSink/KNetConnectSink.cs @@ -0,0 +1,55 @@ +using Java.Util; +using MASES.KNet.Common.Config; +using MASES.KNet.Connect; +using MASES.KNet.Connect.Sink; + +namespace MASES.KNetTemplate.KNetConnect +{ + public class KNetConnectSink : KNetSinkConnector + { + public override string ConnectorName => "MASES.KNetTemplate.KNetConnect.KNetConnectSink"; + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override void TaskConfigs(int index, Map config) + { + + } + + public override Config Validate(Map connectorConfigs) + { + return null; + } + } + + public class KnetConnectSinkTask : KNetSinkTask + { + public override void Put(Collection collection) + { + + } + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override string Version() + { + return "KnetConnectSinkTask"; + } + } +} diff --git a/src/net/templates/templates/knetConnectSink/knetConnectSink.csproj b/src/net/templates/templates/knetConnectSink/knetConnectSink.csproj new file mode 100644 index 0000000000..2360c103aa --- /dev/null +++ b/src/net/templates/templates/knetConnectSink/knetConnectSink.csproj @@ -0,0 +1,16 @@ + + + net6.0;net5.0;netcoreapp3.1;net461 + + + + + + + + + + + + + diff --git a/src/net/templates/templates/knetConnectSource/.template.config/template.json b/src/net/templates/templates/knetConnectSource/.template.config/template.json new file mode 100644 index 0000000000..ab758e4f54 --- /dev/null +++ b/src/net/templates/templates/knetConnectSource/.template.config/template.json @@ -0,0 +1,12 @@ +{ + "$schema": "http://json.schemastore.org/template", + "author": "MASES s.r.l.", + "classifications": [ "Common", "Library"], + "identity": "MASES.KNetTemplate.KNetConnect", + "name": "Library templates: KNet Connect Source project", + "shortName": "knetConnectSource", + "tags": { + "language": "C#", + "type": "project" + } +} \ No newline at end of file diff --git a/src/net/templates/templates/knetConnectSource/KNetConnectSource.cs b/src/net/templates/templates/knetConnectSource/KNetConnectSource.cs new file mode 100644 index 0000000000..450f94c4a6 --- /dev/null +++ b/src/net/templates/templates/knetConnectSource/KNetConnectSource.cs @@ -0,0 +1,55 @@ +using Java.Util; +using MASES.KNet.Common.Config; +using MASES.KNet.Connect; +using MASES.KNet.Connect.Source; + +namespace MASES.KNetTemplate.KNetConnect +{ + public class KNetConnectSource : KNetSourceConnector + { + public override string ConnectorName => "MASES.KNetTemplate.KNetConnect.KNetConnectSource"; + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override void TaskConfigs(int index, Map config) + { + + } + + public override Config Validate(Map connectorConfigs) + { + return null; + } + } + + public class KnetConnectSourceTask : KNetSourceTask + { + public override List Poll() + { + return null; + } + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override string Version() + { + return "KnetConnectSourceTask"; + } + } +} diff --git a/src/net/templates/templates/knetConnectSource/knetConnectSource.csproj b/src/net/templates/templates/knetConnectSource/knetConnectSource.csproj new file mode 100644 index 0000000000..2360c103aa --- /dev/null +++ b/src/net/templates/templates/knetConnectSource/knetConnectSource.csproj @@ -0,0 +1,16 @@ + + + net6.0;net5.0;netcoreapp3.1;net461 + + + + + + + + + + + + + diff --git a/src/net/templates/templates/knetConsumerApp/knetConsumerApp.csproj b/src/net/templates/templates/knetConsumerApp/knetConsumerApp.csproj index a5c08c9cc3..56fa0e88f8 100644 --- a/src/net/templates/templates/knetConsumerApp/knetConsumerApp.csproj +++ b/src/net/templates/templates/knetConsumerApp/knetConsumerApp.csproj @@ -12,6 +12,6 @@ - + diff --git a/src/net/templates/templates/knetPipeStreamApp/knetPipeStreamApp.csproj b/src/net/templates/templates/knetPipeStreamApp/knetPipeStreamApp.csproj index a5c08c9cc3..56fa0e88f8 100644 --- a/src/net/templates/templates/knetPipeStreamApp/knetPipeStreamApp.csproj +++ b/src/net/templates/templates/knetPipeStreamApp/knetPipeStreamApp.csproj @@ -12,6 +12,6 @@ - + diff --git a/src/net/templates/templates/knetProducerApp/knetProducerApp.csproj b/src/net/templates/templates/knetProducerApp/knetProducerApp.csproj index a5c08c9cc3..56fa0e88f8 100644 --- a/src/net/templates/templates/knetProducerApp/knetProducerApp.csproj +++ b/src/net/templates/templates/knetProducerApp/knetProducerApp.csproj @@ -12,6 +12,6 @@ - + diff --git a/tests/KNetBenchmark/KNetBenchmark.csproj b/tests/KNetBenchmark/KNetBenchmark.csproj index e4e7692f79..0cdd32e18b 100644 --- a/tests/KNetBenchmark/KNetBenchmark.csproj +++ b/tests/KNetBenchmark/KNetBenchmark.csproj @@ -8,7 +8,7 @@ Copyright © MASES s.r.l. 2022 MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 net461;netcoreapp3.1;net5.0;net6.0 ..\..\bin\ latest diff --git a/tests/KNetConnectTest/KNetConnectTest.csproj b/tests/KNetConnectTest/KNetConnectTest.csproj new file mode 100644 index 0000000000..06b918ddf8 --- /dev/null +++ b/tests/KNetConnectTest/KNetConnectTest.csproj @@ -0,0 +1,21 @@ + + + KNetConnectTest + MASES.KNetConnectTest + KNetConnectTest - a test tool for KNet Connect + KNetConnectTest - a test tool for KNet Connect + Copyright © MASES s.r.l. 2022 + MASES s.r.l. + MASES s.r.l. + 1.3.0.0 + net461;netcoreapp3.1;net5.0;net6.0 + ..\..\bin\ + latest + + + + + + + + diff --git a/tests/KNetConnectTest/KnetConnectSink.cs b/tests/KNetConnectTest/KnetConnectSink.cs new file mode 100644 index 0000000000..014c46c69b --- /dev/null +++ b/tests/KNetConnectTest/KnetConnectSink.cs @@ -0,0 +1,73 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using Java.Util; +using MASES.KNet.Common.Config; +using MASES.KNet.Connect; +using MASES.KNet.Connect.Sink; + +namespace MASES.KNetConnectTest +{ + public class KnetSinkTestConnector : KNetSinkConnector + { + public override string ConnectorName => "MASES.KNetConnectTest.KnetSinkTestConnector"; + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override void TaskConfigs(int index, Map config) + { + + } + + public override Config Validate(Map connectorConfigs) + { + return null; + } + } + + public class KnetSinkTestTask : KNetSinkTask + { + public override void Put(Collection collection) + { + + } + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override string Version() + { + return "KnetSinkTestTask"; + } + } +} diff --git a/tests/KNetConnectTest/KnetConnectSource.cs b/tests/KNetConnectTest/KnetConnectSource.cs new file mode 100644 index 0000000000..4b98c3699d --- /dev/null +++ b/tests/KNetConnectTest/KnetConnectSource.cs @@ -0,0 +1,73 @@ +/* +* Copyright 2022 MASES s.r.l. +* +* Licensed 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. +* +* Refer to LICENSE for more information. +*/ + +using Java.Util; +using MASES.KNet.Common.Config; +using MASES.KNet.Connect; +using MASES.KNet.Connect.Source; + +namespace MASES.KNetConnectTest +{ + public class KnetSourceTestConnector : KNetSourceConnector + { + public override string ConnectorName => "MASES.KNetConnectTest.KnetSourceTestConnector"; + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override void TaskConfigs(int index, Map config) + { + + } + + public override Config Validate(Map connectorConfigs) + { + return null; + } + } + + public class KnetSourceTestTask : KNetSourceTask + { + public override List Poll() + { + return null; + } + + public override void Start(Map props) + { + + } + + public override void Stop() + { + + } + + public override string Version() + { + return "KnetSourceTestTask"; + } + } +} diff --git a/tests/KNetTest.sln b/tests/KNetTest.sln index a685950184..372266d489 100644 --- a/tests/KNetTest.sln +++ b/tests/KNetTest.sln @@ -19,6 +19,10 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Benchmarks", "Benchmarks", EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tests", "Tests", "{A77F1B9E-5576-46F0-BE58-C85FA9FA3A58}" EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Connect", "Connect", "{6421F571-315C-47BB-A9F9-4542759E968E}" +EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "KNetConnectTest", "KNetConnectTest\KNetConnectTest.csproj", "{7F3A2CE5-ADEE-426F-BE99-D899D87EFF2F}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -49,6 +53,10 @@ Global {C6A11273-A560-4BB1-9234-961D30824DAB}.Debug|Any CPU.Build.0 = Debug|Any CPU {C6A11273-A560-4BB1-9234-961D30824DAB}.Release|Any CPU.ActiveCfg = Release|Any CPU {C6A11273-A560-4BB1-9234-961D30824DAB}.Release|Any CPU.Build.0 = Release|Any CPU + {7F3A2CE5-ADEE-426F-BE99-D899D87EFF2F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {7F3A2CE5-ADEE-426F-BE99-D899D87EFF2F}.Debug|Any CPU.Build.0 = Debug|Any CPU + {7F3A2CE5-ADEE-426F-BE99-D899D87EFF2F}.Release|Any CPU.ActiveCfg = Release|Any CPU + {7F3A2CE5-ADEE-426F-BE99-D899D87EFF2F}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -59,6 +67,7 @@ Global {DE8D288D-8438-42EA-892F-F59840CFE322} = {A77F1B9E-5576-46F0-BE58-C85FA9FA3A58} {8A6C9438-A389-4563-98EE-1386E0E62AAA} = {14753FBA-8F9B-48A1-9983-7DC37CB32914} {C6A11273-A560-4BB1-9234-961D30824DAB} = {14753FBA-8F9B-48A1-9983-7DC37CB32914} + {7F3A2CE5-ADEE-426F-BE99-D899D87EFF2F} = {6421F571-315C-47BB-A9F9-4542759E968E} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {0A7C16DC-1BAA-44BC-AA1C-D40B7B61878E} diff --git a/tests/KNetTest/KNetTest.csproj b/tests/KNetTest/KNetTest.csproj index 228b7477a9..575f761c09 100644 --- a/tests/KNetTest/KNetTest.csproj +++ b/tests/KNetTest/KNetTest.csproj @@ -8,7 +8,7 @@ Copyright © MASES s.r.l. 2022 MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 net461;netcoreapp3.1;net5.0;net6.0 ..\..\bin\ latest diff --git a/tests/KNetTestAdmin/KNetTestAdmin.csproj b/tests/KNetTestAdmin/KNetTestAdmin.csproj index bf0c9bc40e..cda9bd9d2c 100644 --- a/tests/KNetTestAdmin/KNetTestAdmin.csproj +++ b/tests/KNetTestAdmin/KNetTestAdmin.csproj @@ -8,7 +8,7 @@ Copyright © MASES s.r.l. 2022 MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 net461;netcoreapp3.1;net5.0;net6.0 ..\..\bin\ latest diff --git a/tests/KNetTestStreams/KNetTestStreams.csproj b/tests/KNetTestStreams/KNetTestStreams.csproj index 5991070b9b..8e10e80762 100644 --- a/tests/KNetTestStreams/KNetTestStreams.csproj +++ b/tests/KNetTestStreams/KNetTestStreams.csproj @@ -8,7 +8,7 @@ Copyright © MASES s.r.l. 2022 MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 net461;netcoreapp3.1;net5.0;net6.0 ..\..\bin\ latest diff --git a/tests/KNetTopicCopyBenchmark/KNetTopicCopyBenchmark.csproj b/tests/KNetTopicCopyBenchmark/KNetTopicCopyBenchmark.csproj index 5f06b581fe..c23c75aeb2 100644 --- a/tests/KNetTopicCopyBenchmark/KNetTopicCopyBenchmark.csproj +++ b/tests/KNetTopicCopyBenchmark/KNetTopicCopyBenchmark.csproj @@ -8,7 +8,7 @@ Copyright © MASES s.r.l. 2022 MASES s.r.l. MASES s.r.l. - 1.2.4.0 + 1.3.0.0 net461;netcoreapp3.1;net5.0;net6.0 ..\..\bin\ latest