Added sample code and script for RSI with Virtual Threads and ATP

java/rsi-atp-vt/README.md

@@ -0,0 +1,2 @@
+## Java - Oracle Developers on Medium.com
+[High-throughput stream processing with the Java Library for Reactive Streams Ingestion (RSI), Virtual Threads, and the Oracle ATP Database](https://medium.com/oracledevs/high-throughput-stream-processing-with-the-java-library-for-reactive-streams-ingestion-rsi-8aa0fc9d167)

java/rsi-atp-vt/pom.xml

@@ -0,0 +1,121 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.oracle.dev.jdbc</groupId>
+	<artifactId>rsi-atp-vt</artifactId>
+	<version>1.0-SNAPSHOT</version>
+
+	<name>rsi-atp-vt</name>
+	<description>A simple rsi-atp-vt.</description>	
+	<url>https://github.com/juarezjuniorgithub/rsi-atp-vt</url>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<maven.compiler.source>19</maven.compiler.source>
+		<maven.compiler.target>19</maven.compiler.target>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<version>3.8.1</version>
+		</dependency>
+		<!-- Oracle JDBC JARs -->
+		<dependency>
+			<groupId>com.oracle.database.jdbc</groupId>
+			<artifactId>rsi</artifactId>
+			<version>21.6.0.0.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.oracle.database.jdbc</groupId>
+			<artifactId>ojdbc11</artifactId>
+			<version>21.6.0.0.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.oracle.database.jdbc</groupId>
+			<artifactId>ucp11</artifactId>
+			<version>21.6.0.0.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.oracle.ojdbc</groupId>
+			<artifactId>ons</artifactId>
+			<version>19.3.0.0</version>
+		</dependency>
+		<dependency>
+			<groupId>com.oracle.database.jdbc</groupId>
+			<artifactId>oraclepki</artifactId>
+			<version>19.11.0.0</version>
+		</dependency>
+		<dependency>
+			<groupId>com.oracle.database.security</groupId>
+			<artifactId>osdt_core</artifactId>
+			<version>21.6.0.0.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.oracle.database.security</groupId>
+			<artifactId>osdt_cert</artifactId>
+			<version>21.6.0.0.1</version>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<pluginManagement>
+			<!-- lock down plugins versions to avoid using Maven defaults (may be moved to parent pom) -->
+			<plugins>
+				<plugin>
+					<artifactId>maven-clean-plugin</artifactId>
+					<version>3.1.0</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-site-plugin</artifactId>
+					<version>3.7.1</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-project-info-reports-plugin</artifactId>
+					<version>3.0.0</version>
+				</plugin>
+				<!-- see http://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging -->
+				<plugin>
+					<artifactId>maven-resources-plugin</artifactId>
+					<version>3.0.2</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-compiler-plugin</artifactId>
+					<version>3.8.0</version>
+					<configuration>
+						<compilerArgs>
+							<arg>--enable-preview</arg>
+						</compilerArgs>
+					</configuration>
+				</plugin>
+				<plugin>
+					<artifactId>maven-surefire-plugin</artifactId>
+					<version>2.22.1</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-jar-plugin</artifactId>
+					<version>3.0.2</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-install-plugin</artifactId>
+					<version>2.5.2</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-deploy-plugin</artifactId>
+					<version>2.8.2</version>
+				</plugin>
+			</plugins>
+		</pluginManagement>
+	</build>
+
+	<reporting>
+		<plugins>
+			<plugin>
+				<artifactId>maven-project-info-reports-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</reporting>
+</project>

java/rsi-atp-vt/script/rsi.sql

@@ -0,0 +1,9 @@
+CREATE USER RSIATP_USER IDENTIFIED BY "<PASSWORD>";
+GRANT DWROLE TO RSIATP_USER;
+GRANT UNLIMITED TABLESPACE TO RSIATP_USER;
+
+CREATE TABLE RSIATP_USER.CUSTOMERS (
+ ID NUMBER(20,0),
+ NAME VARCHAR2(20 BYTE),
+ REGION VARCHAR2(20 BYTE)
+);

java/rsi-atp-vt/src/main/java/com/oracle/dev/jdbc/Customer.java

@@ -0,0 +1,48 @@
+/*
+  Copyright (c) 2021, 2022, Oracle and/or its affiliates.
+
+  This software is dual-licensed to you under the Universal Permissive License
+  (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License
+  2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+  either license.
+
+  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
+
+     https://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 com.oracle.dev.jdbc;
+
+import oracle.rsi.StreamEntity;
+import oracle.rsi.StreamField;
+
+@StreamEntity(tableName = "customers")
+public class Customer {
+
+	public Customer(long id, String name, String region) {
+		super();
+		this.id = id;
+		this.name = name;
+		this.region = region;
+	}
+
+	@StreamField
+	public long id;
+
+	@StreamField
+	public String name;
+
+	@StreamField(columnName = "region")
+	public String region;
+
+	String someRandomField;
+
+}

java/rsi-atp-vt/src/main/java/com/oracle/dev/jdbc/DatabaseConfig.java

@@ -0,0 +1,72 @@
+/*
+  Copyright (c) 2021, 2022, Oracle and/or its affiliates.
+
+  This software is dual-licensed to you under the Universal Permissive License
+  (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License
+  2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+  either license.
+
+  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
+
+     https://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 com.oracle.dev.jdbc;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Properties;
+
+/**
+ * <p>
+ * Configuration for connecting code samples to an Oracle Database instance.
+ * </p>
+ */
+public class DatabaseConfig {
+
+	private static final Properties CONFIG = new Properties();
+
+	static {
+		try {
+			var fileStream = Files
+					.newInputStream(Path.of("C:\\java-projects\\rsi-atp-vt\\src\\main\\resources\\config.properties"));
+			CONFIG.load(fileStream);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+	}
+
+	private static final String DB_USER = CONFIG.getProperty("DB_USER");
+
+	private static final String DB_URL = CONFIG.getProperty("DB_URL");
+
+	private static final String DB_PASSWORD = CONFIG.getProperty("DB_PASSWORD");
+
+	private static final String DB_SCHEMA = CONFIG.getProperty("DB_SCHEMA");
+
+	public static String getDbUser() {
+		return DB_USER;
+	}
+
+	public static String getDbUrl() {
+		return DB_URL;
+	}
+
+	public static String getDbPassword() {
+		return DB_PASSWORD;
+	}
+
+	public static String getDbSchema() {
+		return DB_SCHEMA;
+	}
+
+}

java/rsi-atp-vt/src/main/java/com/oracle/dev/jdbc/RSIApp.java

@@ -0,0 +1,84 @@
+/*
+  Copyright (c) 2021, 2022, Oracle and/or its affiliates.
+
+  This software is dual-licensed to you under the Universal Permissive License
+  (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License
+  2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+  either license.
+
+  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
+
+     https://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 com.oracle.dev.jdbc;
+
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.IntStream;
+
+import oracle.rsi.PushPublisher;
+import oracle.rsi.ReactiveStreamsIngestion;
+
+public class RSIApp {
+
+	private ExecutorService workerThreadPool;
+	private ReactiveStreamsIngestion rsi;
+	private PushPublisher<Object[]> publisher;
+
+	public static void main(String[] args) throws SQLException {
+		RSIApp app = new RSIApp();
+		app.init();
+		app.publish();
+	}
+
+	private void init() {
+		// virtual threads
+		workerThreadPool = Executors.newVirtualThreadPerTaskExecutor();
+		rsi = ReactiveStreamsIngestion.builder().url(DatabaseConfig.getDbUrl()).username(DatabaseConfig.getDbUser())
+				.password(DatabaseConfig.getDbPassword()).schema(DatabaseConfig.getDbSchema())
+				.executor(workerThreadPool).bufferRows(200).bufferInterval(Duration.ofSeconds(20))
+				.entity(Customer.class).build();
+	}
+
+	private void publish() {
+		publisher = ReactiveStreamsIngestion.pushPublisher();
+		publisher.subscribe(rsi.subscriber());
+		// virtual threads
+		var threads = IntStream.range(0, 150).mapToObj(i -> Thread.startVirtualThread(() -> {
+			publisher.accept(new Object[] { ThreadLocalRandom.current().nextLong(), "Duke Java", "West" });
+		})).toList();
+
+		for (var thread : threads) {
+			try {
+				thread.join();
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+		destroy();
+	}
+
+	private void destroy() {
+		try {
+			publisher.close();
+			rsi.close();
+		} catch (Exception e) {
+			e.printStackTrace();
+		} finally {
+			workerThreadPool.shutdown();
+		}
+	}
+
+}

java/rsi-atp-vt/src/main/resources/config.properties

@@ -0,0 +1,6 @@
+# https://docs.oracle.com/en/cloud/paas/autonomous-database/adbsa/connect-jdbc-thin-wallet.html
+# jdbc:oracle:thin:@dbname_medium?TNS_ADMIN=<PATH_TO_YOUR_WALLET>
+DB_URL=jdbc:oracle:thin:@rsitestdb_medium?TNS_ADMIN=<PATH_TO_YOUR_WALLET>
+DB_USER=<DB_USER>
+DB_PASSWORD=<DB_PASSWORD>
+DB_SCHEMA=<DB_SCHEMA>

java/rsi-atp-vt/terraform/adb.tf

@@ -0,0 +1,16 @@
+resource "oci_database_autonomous_database" "autonomous_database" {
+  compartment_id              = var.compartment_ocid
+  db_name                     = upper(replace(var.name, "-", ""))
+  display_name                = format("%s%s", "adb-", var.name)
+  db_workload                 = "OLTP"
+  is_dedicated                = false
+  is_free_tier                = false
+  cpu_core_count              = 1
+  data_storage_size_in_tbs    = 1
+  admin_password              = var.adb_admin_password
+  is_mtls_connection_required = true
+  customer_contacts {
+    email = var.adb_customer_contacts_email
+  }
+
+}

java/rsi-atp-vt/terraform/config.tf

@@ -0,0 +1,7 @@
+terraform {
+  required_providers {
+    oci = {
+      source = "oracle/oci"
+    }
+  }
+}

java/rsi-atp-vt/terraform/outputs.tf

@@ -0,0 +1,3 @@
+output "adb_connection_strings" {
+  value = oci_database_autonomous_database.autonomous_database.connection_strings
+}