This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 6b5bd1377 (InfluxDB, HDFS, JMS) Replace java.lang.reflect with
VarHandle/MethodHandle (#1752)
6b5bd1377 is described below
commit 6b5bd13771650e1aa41947cd8aaf2f34a227dd73
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jul 14 00:56:34 2026 +0800
(InfluxDB, HDFS, JMS) Replace java.lang.reflect with VarHandle/MethodHandle
(#1752)
* refactor: replace java.lang.reflect with VarHandle/MethodHandle in
InfluxDB, HDFS, JMS
Motivation:
java.lang.reflect Field.get/set and Constructor.newInstance bypass JIT
inlining, adding overhead in hot paths like InfluxDB result mapping,
HDFS configuration, and JMS connection factory instantiation.
Modification:
- InfluxDB: replace Field.get/set with VarHandle for result mapping;
hoist privateLookupIn per class; use invokeWithArguments()
- HDFS: replace Constructor.newInstance with MethodHandle for
configuration class instantiation
- JMS: replace Constructor.newInstance with cached MethodHandle for
ConnectionFactory creation
Result:
JIT-inlinable field access and constructor calls in connector hot paths.
Refs #1752
* perf: use invoke for zero-arg method handles
Motivation:\nAvoid the allocation-heavy invokeWithArguments() path for
no-arg constructor handles in hot connector code.\n\nModification:\nUse
MethodHandle.invoke() for HdfsSource sequence element construction and InfluxDB
row instantiation.\n\nResult:\nConnector reflection bridges keep the
MethodHandle migration but remove unnecessary generic invocation overhead.
---
.../connectors/hdfs/scaladsl/HdfsSource.scala | 23 ++-
.../impl/PekkoConnectorsResultMapperHelper.scala | 195 ++++++++++++---------
.../test/java/docs/javadsl/InfluxDbSourceTest.java | 2 +-
.../src/test/java/docs/javadsl/InfluxDbTest.java | 2 +-
influxdb/src/test/java/docs/javadsl/TestUtils.java | 26 ++-
.../PekkoConnectorsResultMapperHelperSpec.scala | 73 ++++++++
.../scala/docs/scaladsl/JmsConnectorsSpec.scala | 19 +-
7 files changed, 237 insertions(+), 103 deletions(-)
diff --git
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala
index 53cc8cfee..1cccca768 100644
---
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala
+++
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala
@@ -13,6 +13,9 @@
package org.apache.pekko.stream.connectors.hdfs.scaladsl
+import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType }
+import java.util.concurrent.ConcurrentHashMap
+
import org.apache.pekko
import pekko.NotUsed
import pekko.stream.ActorAttributes.IODispatcher
@@ -27,6 +30,20 @@ import scala.concurrent.Future
object HdfsSource {
+ private val constructorCache = new ConcurrentHashMap[Class[?],
MethodHandle]()
+ private val LOOKUP = MethodHandles.lookup()
+
+ private def getNoArgConstructor[T](clazz: Class[T]): MethodHandle = {
+ var handle = constructorCache.get(clazz)
+ if (handle == null) {
+ val lookup = MethodHandles.privateLookupIn(clazz, LOOKUP)
+ handle = lookup.findConstructor(clazz, MethodType.methodType(Void.TYPE))
+ val existing = constructorCache.putIfAbsent(clazz, handle)
+ if (existing != null) handle = existing
+ }
+ handle
+ }
+
/**
* Scala API: creates a `Source` that consumes as `ByteString`
*
@@ -69,10 +86,12 @@ object HdfsSource {
classK: Class[K],
classV: Class[V]): Source[(K, V), NotUsed] = {
val reader: SequenceFile.Reader = new SequenceFile.Reader(fs.getConf,
SequenceFile.Reader.file(path))
+ val keyHandle = getNoArgConstructor(classK)
+ val valueHandle = getNoArgConstructor(classV)
val it = Iterator
.continually {
- val key = classK.getDeclaredConstructor().newInstance()
- val value = classV.getDeclaredConstructor().newInstance()
+ val key = keyHandle.invoke().asInstanceOf[K]
+ val value = valueHandle.invoke().asInstanceOf[V]
val hasCurrent = reader.next(key, value)
(hasCurrent, (key, value))
}
diff --git
a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala
b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala
index 681ed63dd..81efbfe62 100644
---
a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala
+++
b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala
@@ -13,10 +13,11 @@
package org.apache.pekko.stream.connectors.influxdb.impl
-import java.lang.reflect.Field
+import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType, VarHandle }
import java.time.Instant
import java.time.format.DateTimeFormatterBuilder
import java.time.temporal.ChronoField
+import java.util.function.Function
import java.util.concurrent.{ ConcurrentHashMap, ConcurrentMap }
import java.util.concurrent.TimeUnit
@@ -35,7 +36,7 @@ import scala.jdk.CollectionConverters._
@InternalApi
private[impl] class PekkoConnectorsResultMapperHelper {
- val CLASS_FIELD_CACHE: ConcurrentHashMap[String, ConcurrentMap[String,
Field]] = new ConcurrentHashMap();
+ import PekkoConnectorsResultMapperHelper.{ CLASS_COLUMN_CACHE,
CONSTRUCTOR_CACHE, ColumnInfo, LOOKUP }
private val FRACTION_MIN_WIDTH = 0
private val FRACTION_MAX_WIDTH = 9
@@ -57,7 +58,7 @@ private[impl] class PekkoConnectorsResultMapperHelper {
throwExceptionIfMissingAnnotation(model.getClass)
cacheClassFields(model.getClass)
- val colNameAndFieldMap: ConcurrentMap[String, Field] =
CLASS_FIELD_CACHE.get(model.getClass.getName)
+ val colMap: ConcurrentMap[String, ColumnInfo] =
CLASS_COLUMN_CACHE.get(model.getClass)
try {
val modelType = model.getClass();
@@ -66,27 +67,18 @@ private[impl] class PekkoConnectorsResultMapperHelper {
val time = timeUnit.convert(System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
val pointBuilder: Point.Builder =
Point.measurement(measurement).time(time, timeUnit);
- for (key <- colNameAndFieldMap.keySet().asScala) {
- val field = colNameAndFieldMap.get(key)
- val column = field.getAnnotation(classOf[Column])
- val columnName: String = column.name()
- val fieldType: Class[?] = field.getType()
+ for (key <- colMap.keySet().asScala) {
+ val colInfo = colMap.get(key)
+ val value = colInfo.varHandle.get(model);
- val isAccessible = field.isAccessible()
- if (!isAccessible) {
- field.setAccessible(true);
- }
-
- val value = field.get(model);
-
- if (column.tag()) {
- pointBuilder.tag(columnName, value.toString());
- } else if ("time".equals(columnName)) {
+ if (colInfo.isTag) {
+ pointBuilder.tag(colInfo.columnName, value.toString());
+ } else if ("time".equals(colInfo.columnName)) {
if (value != null) {
- setTime(pointBuilder, fieldType, timeUnit, value);
+ setTime(pointBuilder, colInfo.fieldType, timeUnit, value);
}
} else {
- setField(pointBuilder, fieldType, columnName, value);
+ setField(pointBuilder, colInfo.fieldType, colInfo.columnName, value);
}
}
@@ -96,27 +88,13 @@ private[impl] class PekkoConnectorsResultMapperHelper {
}
}
- private[impl] def cacheClassFields(clazz: Class[?]) =
- if (!CLASS_FIELD_CACHE.containsKey(clazz.getName)) {
- val initialMap: ConcurrentMap[String, Field] = new ConcurrentHashMap()
- var influxColumnAndFieldMap =
CLASS_FIELD_CACHE.putIfAbsent(clazz.getName, initialMap)
-
- if (influxColumnAndFieldMap == null) {
- influxColumnAndFieldMap = initialMap;
- }
-
- var c = clazz;
-
- while (c != null) {
- for (field <- c.getDeclaredFields()) {
- val colAnnotation = field.getAnnotation(classOf[Column]);
- if (colAnnotation != null) {
- influxColumnAndFieldMap.put(colAnnotation.name(), field);
- }
- }
- c = c.getSuperclass();
- }
- }
+ private[impl] def cacheClassFields(clazz: Class[?]): Unit =
+ CLASS_COLUMN_CACHE.computeIfAbsent(
+ clazz,
+ new Function[Class[?], ConcurrentMap[String, ColumnInfo]] {
+ override def apply(clazz: Class[?]): ConcurrentMap[String, ColumnInfo]
=
+ PekkoConnectorsResultMapperHelper.classColumnMap(clazz)
+ })
private[impl] def parseSeriesAs[T](clazz: Class[T], series:
QueryResult.Series, precision: TimeUnit): List[T] = {
cacheClassFields(clazz)
@@ -155,111 +133,115 @@ private[impl] class PekkoConnectorsResultMapperHelper {
throw new IllegalArgumentException(
"Class " + clazz.getName + " is not annotated with @" +
classOf[Measurement].getSimpleName)
+ private def getOrCreateConstructorHandle(clazz: Class[?]): MethodHandle =
+ CONSTRUCTOR_CACHE.computeIfAbsent(
+ clazz,
+ new Function[Class[?], MethodHandle] {
+ override def apply(clazz: Class[?]): MethodHandle =
+ MethodHandles.privateLookupIn(clazz, LOOKUP).findConstructor(clazz,
MethodType.methodType(Void.TYPE))
+ })
+
private def parseRowAs[T](clazz: Class[T],
columns: java.util.List[String],
values: java.util.List[AnyRef],
precision: TimeUnit): T =
try {
- val fieldMap = CLASS_FIELD_CACHE.get(clazz.getName)
+ val colMap = CLASS_COLUMN_CACHE.get(clazz)
- val obj: T = clazz.getDeclaredConstructor().newInstance()
+ val handle = getOrCreateConstructorHandle(clazz)
+ val obj: T = handle.invoke().asInstanceOf[T]
for (i <- 0 until columns.size()) {
- val correspondingField = fieldMap.get(columns.get(i))
- if (correspondingField != null) {
- setFieldValue(obj, correspondingField, values.get(i), precision)
+ val colInfo = colMap.get(columns.get(i))
+ if (colInfo != null) {
+ setFieldValue(obj, colInfo, values.get(i), precision)
}
}
obj
} catch {
- case e @ (_: InstantiationException | _: IllegalAccessException) =>
+ case e: InfluxDBMapperException => throw e
+ case e: Exception =>
throw new InfluxDBMapperException(e)
}
- @throws[IllegalArgumentException]
- @throws[IllegalAccessException]
- private def setFieldValue[T](obj: T, field: Field, value: Any, precision:
TimeUnit): Unit = {
+ private def setFieldValue[T](obj: T, colInfo: ColumnInfo, value: Any,
precision: TimeUnit): Unit = {
if (value == null) return
- val fieldType = field.getType
+ val fieldType = colInfo.fieldType
try {
- val isAccessible = field.isAccessible()
- if (!isAccessible) field.setAccessible(true)
- if (fieldValueModified(fieldType, field, obj, value, precision) ||
fieldValueForPrimitivesModified(
- fieldType,
- field,
- obj,
- value) || fieldValueForPrimitiveWrappersModified(fieldType, field,
obj, value)) return
+ if (fieldValueModified(fieldType, colInfo, obj, value, precision) ||
+ fieldValueForPrimitivesModified(fieldType, colInfo, obj, value) ||
+ fieldValueForPrimitiveWrappersModified(fieldType, colInfo, obj,
value)) return
val msg =
- s"""Class '${obj.getClass.getName}' field '${field.getName}' is from
an unsupported type '${field.getType}'."""
+ s"""Class '${obj.getClass.getName}' field '${colInfo.fieldName}' is
from an unsupported type '${colInfo.fieldType}'."""
throw new InfluxDBMapperException(msg)
} catch {
case e: ClassCastException =>
val msg =
- s"""Class '${obj.getClass.getName}' field '${field.getName}' was
defined with a different field type and caused a ClassCastException.
+ s"""Class '${obj.getClass.getName}' field '${colInfo.fieldName}' was
defined with a different field type and caused a ClassCastException.
|The correct type is '${value.getClass.getName}' (current field
value: '${value}')""".stripMargin
throw new InfluxDBMapperException(msg)
}
}
- @throws[IllegalArgumentException]
- @throws[IllegalAccessException]
- private def fieldValueForPrimitivesModified[T](fieldType: Class[?], field:
Field, obj: T, value: Any): Boolean =
+ private def fieldValueForPrimitivesModified[T](
+ fieldType: Class[?],
+ colInfo: ColumnInfo,
+ obj: T,
+ value: Any): Boolean =
if (classOf[Double].isAssignableFrom(fieldType)) {
- field.setDouble(obj, value.asInstanceOf[Double].doubleValue)
+ colInfo.varHandle.set(obj, value.asInstanceOf[Double].doubleValue)
true
} else if (classOf[Long].isAssignableFrom(fieldType)) {
- field.setLong(obj, value.asInstanceOf[Double].longValue)
+ colInfo.varHandle.set(obj, value.asInstanceOf[Double].longValue)
true
} else if (classOf[Int].isAssignableFrom(fieldType)) {
- field.setInt(obj, value.asInstanceOf[Double].intValue)
+ colInfo.varHandle.set(obj, value.asInstanceOf[Double].intValue)
true
} else if (classOf[Boolean].isAssignableFrom(fieldType)) {
- field.setBoolean(obj, String.valueOf(value).toBoolean)
+ colInfo.varHandle.set(obj, String.valueOf(value).toBoolean)
true
} else {
false
}
- @throws[IllegalArgumentException]
- @throws[IllegalAccessException]
- private def fieldValueForPrimitiveWrappersModified[T](fieldType: Class[?],
- field: Field,
+ private def fieldValueForPrimitiveWrappersModified[T](
+ fieldType: Class[?],
+ colInfo: ColumnInfo,
obj: T,
value: Any): Boolean =
if (classOf[java.lang.Double].isAssignableFrom(fieldType)) {
- field.set(obj, value)
+ colInfo.varHandle.set(obj, value)
true
} else if (classOf[java.lang.Long].isAssignableFrom(fieldType)) {
- field.set(obj, value.asInstanceOf[Double].longValue())
+ colInfo.varHandle.set(obj, value.asInstanceOf[Double].longValue())
true
} else if (classOf[Integer].isAssignableFrom(fieldType)) {
- field.set(obj, value.asInstanceOf[java.lang.Integer])
+ colInfo.varHandle.set(obj, value.asInstanceOf[java.lang.Integer])
true
} else if (classOf[java.lang.Boolean].isAssignableFrom(fieldType)) {
- field.set(obj, value.asInstanceOf[java.lang.Boolean])
+ colInfo.varHandle.set(obj, value.asInstanceOf[java.lang.Boolean])
true
} else {
false
}
- @throws[IllegalArgumentException]
- @throws[IllegalAccessException]
- private def fieldValueModified[T](fieldType: Class[?],
- field: Field,
+ private def fieldValueModified[T](
+ fieldType: Class[?],
+ colInfo: ColumnInfo,
obj: T,
value: Any,
precision: TimeUnit): Boolean =
if (classOf[String].isAssignableFrom(fieldType)) {
- field.set(obj, String.valueOf(value))
+ colInfo.varHandle.set(obj, String.valueOf(value))
true
} else if (classOf[Instant].isAssignableFrom(fieldType)) {
- val instant: Instant = getInstant(field, value, precision)
- field.set(obj, instant)
+ val instant: Instant = getInstant(colInfo, value, precision)
+ colInfo.varHandle.set(obj, instant)
true
} else {
false
}
- private def getInstant(field: Field, value: Any, precision: TimeUnit):
Instant =
+ private def getInstant(colInfo: ColumnInfo, value: Any, precision:
TimeUnit): Instant =
if (value.isInstanceOf[String])
Instant.from(RFC3339_FORMATTER.parse(String.valueOf(value)))
else if (value.isInstanceOf[java.lang.Long])
Instant.ofEpochMilli(toMillis(value.asInstanceOf[Long], precision))
else if (value.isInstanceOf[java.lang.Double])
@@ -267,9 +249,54 @@ private[impl] class PekkoConnectorsResultMapperHelper {
else if (value.isInstanceOf[java.lang.Integer])
Instant.ofEpochMilli(toMillis(value.asInstanceOf[Integer].longValue,
precision))
else {
- throw new InfluxDBMapperException(s"""Unsupported type ${field.getClass}
for field ${field.getName}""")
+ throw new InfluxDBMapperException(s"""Unsupported type for field
${colInfo.fieldName}""")
}
private def toMillis(value: Long, precision: TimeUnit) =
TimeUnit.MILLISECONDS.convert(value, precision)
}
+
+@InternalApi
+private[impl] object PekkoConnectorsResultMapperHelper {
+ private val CLASS_COLUMN_CACHE: ConcurrentHashMap[Class[?],
ConcurrentMap[String, ColumnInfo]] =
+ new ConcurrentHashMap()
+
+ private val CONSTRUCTOR_CACHE: ConcurrentHashMap[Class[?], MethodHandle] =
new ConcurrentHashMap()
+
+ private val LOOKUP = MethodHandles.lookup()
+
+ final class ColumnInfo(
+ val columnName: String,
+ val isTag: Boolean,
+ val fieldType: Class[?],
+ val varHandle: VarHandle,
+ val fieldName: String)
+
+ private def classColumnMap(clazz: Class[?]): ConcurrentMap[String,
ColumnInfo] = {
+ val columnMap: ConcurrentMap[String, ColumnInfo] = new ConcurrentHashMap()
+ var c = clazz
+
+ while (c != null) {
+ val fields = c.getDeclaredFields
+ if (fields.nonEmpty) {
+ val privateLookup = MethodHandles.privateLookupIn(c, LOOKUP)
+ for (field <- fields) {
+ val colAnnotation = field.getAnnotation(classOf[Column])
+ if (colAnnotation != null) {
+ val varHandle = privateLookup.findVarHandle(c, field.getName,
field.getType)
+ val colInfo = new ColumnInfo(
+ colAnnotation.name(),
+ colAnnotation.tag(),
+ field.getType,
+ varHandle,
+ field.getName)
+ columnMap.put(colAnnotation.name(), colInfo)
+ }
+ }
+ }
+ c = c.getSuperclass
+ }
+
+ columnMap
+ }
+}
diff --git a/influxdb/src/test/java/docs/javadsl/InfluxDbSourceTest.java
b/influxdb/src/test/java/docs/javadsl/InfluxDbSourceTest.java
index 390f6f44e..88bd96a34 100644
--- a/influxdb/src/test/java/docs/javadsl/InfluxDbSourceTest.java
+++ b/influxdb/src/test/java/docs/javadsl/InfluxDbSourceTest.java
@@ -56,7 +56,7 @@ public class InfluxDbSourceTest {
}
@BeforeEach
- public void setUp() throws Exception {
+ public void setUp() throws Throwable {
populateDatabase(influxDB, InfluxDbSourceCpu.class);
}
diff --git a/influxdb/src/test/java/docs/javadsl/InfluxDbTest.java
b/influxdb/src/test/java/docs/javadsl/InfluxDbTest.java
index ac092183a..a1391f92e 100644
--- a/influxdb/src/test/java/docs/javadsl/InfluxDbTest.java
+++ b/influxdb/src/test/java/docs/javadsl/InfluxDbTest.java
@@ -102,7 +102,7 @@ public class InfluxDbTest {
}
@BeforeEach
- public void setUp() throws Exception {
+ public void setUp() throws Throwable {
populateDatabase(influxDB, InfluxDbCpu.class);
}
diff --git a/influxdb/src/test/java/docs/javadsl/TestUtils.java
b/influxdb/src/test/java/docs/javadsl/TestUtils.java
index fdb453a54..a0bc26c71 100644
--- a/influxdb/src/test/java/docs/javadsl/TestUtils.java
+++ b/influxdb/src/test/java/docs/javadsl/TestUtils.java
@@ -17,7 +17,9 @@ import static docs.javadsl.TestConstants.INFLUXDB_URL;
import static docs.javadsl.TestConstants.PASSWORD;
import static docs.javadsl.TestConstants.USERNAME;
-import java.lang.reflect.Constructor;
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
import java.time.Instant;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -40,17 +42,27 @@ public class TestUtils {
// #init-client
}
- public static void populateDatabase(InfluxDB influxDB, Class<?> clazz)
throws Exception {
+ public static void populateDatabase(InfluxDB influxDB, Class<?> clazz)
throws Throwable {
InfluxDBMapper influxDBMapper = new InfluxDBMapper(influxDB);
- Constructor<?> cons =
- clazz.getConstructor(
- Instant.class, String.class, String.class, Double.class,
Boolean.class, Long.class);
+ MethodHandle cons =
+ MethodHandles.privateLookupIn(clazz, MethodHandles.lookup())
+ .findConstructor(
+ clazz,
+ MethodType.methodType(
+ Void.TYPE,
+ Instant.class,
+ String.class,
+ String.class,
+ Double.class,
+ Boolean.class,
+ Long.class));
Object firstCore =
- cons.newInstance(
+ cons.invokeWithArguments(
Instant.now().minusSeconds(1000), "local_1", "eu-west-2", 1.4d,
true, 123l);
influxDBMapper.save(firstCore);
Object secondCore =
- cons.newInstance(Instant.now().minusSeconds(500), "local_2",
"eu-west-2", 1.4d, true, 123l);
+ cons.invokeWithArguments(
+ Instant.now().minusSeconds(500), "local_2", "eu-west-2", 1.4d,
true, 123l);
influxDBMapper.save(secondCore);
}
diff --git
a/influxdb/src/test/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelperSpec.scala
b/influxdb/src/test/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelperSpec.scala
new file mode 100644
index 000000000..dbda29e7a
--- /dev/null
+++
b/influxdb/src/test/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelperSpec.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.pekko.stream.connectors.influxdb.impl
+
+import java.time.Instant
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+
+import docs.scaladsl.InfluxDbSourceCpu
+import org.influxdb.dto.QueryResult
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class PekkoConnectorsResultMapperHelperSpec extends AnyWordSpec with Matchers {
+
+ "PekkoConnectorsResultMapperHelper" should {
+ "map annotated fields with VarHandles" in {
+ val helper = new PekkoConnectorsResultMapperHelper
+ val series = new QueryResult.Series
+ series.setColumns(List("time", "hostname", "region", "idle",
"happydevop", "uptimesecs").asJava)
+ series.setValues(
+ List(
+ List[AnyRef](
+ "2026-07-06T09:42:00Z",
+ "local_1",
+ "eu-west-2",
+ Double.box(1.5d),
+ java.lang.Boolean.TRUE,
+ Double.box(123d)).asJava).asJava)
+
+ val parsed = helper.parseSeriesAs(classOf[InfluxDbSourceCpu], series,
TimeUnit.MILLISECONDS)
+
+ parsed should have size 1
+ parsed.head.getTime shouldBe Instant.parse("2026-07-06T09:42:00Z")
+ parsed.head.getHostname shouldBe "local_1"
+ parsed.head.getRegion shouldBe "eu-west-2"
+ parsed.head.getIdle shouldBe 1.5d
+ parsed.head.getHappydevop shouldBe true
+ parsed.head.getUptimeSecs shouldBe 123L
+ }
+
+ "convert annotated models to points with cached MethodHandles and
VarHandles" in {
+ val helper = new PekkoConnectorsResultMapperHelper
+ val model =
+ new InfluxDbSourceCpu(Instant.parse("2026-07-06T09:42:00Z"),
"local_2", "eu-west-1", 2.5d, false, 125L)
+
+ val lineProtocol = helper.convertModelToPoint(model).lineProtocol()
+
+ lineProtocol should include("cpu")
+ lineProtocol should include("hostname=local_2")
+ lineProtocol should include("region=eu-west-1")
+ lineProtocol should include("idle=2.5")
+ lineProtocol should include("happydevop=false")
+ lineProtocol should include("uptimesecs=125")
+ }
+ }
+}
diff --git a/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
b/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
index bf9c647a6..2bb356491 100644
--- a/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
+++ b/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
@@ -26,6 +26,7 @@ import org.mockito.Mockito._
import org.mockito.invocation.InvocationOnMock
import org.mockito.stubbing.Answer
+import java.lang.invoke.MethodHandles
import java.nio.charset.StandardCharsets
import java.util.UUID
import java.util.concurrent.atomic.AtomicInteger
@@ -511,11 +512,12 @@ class JmsConnectorsSpec extends JmsSpec {
// make sure connection was closed
eventually {
- val isClosedField =
-
connectionFactory.cachedConnection.getClass.getDeclaredFields.filter(x =>
x.getName == "closed").head
- isClosedField.setAccessible(true)
+ val conn = connectionFactory.cachedConnection
+ val connClass = conn.getClass
+ val lookup = MethodHandles.privateLookupIn(connClass,
MethodHandles.lookup())
+ val isClosedHandle = lookup.findVarHandle(connClass, "closed",
classOf[Boolean])
- val isClosedValue =
isClosedField.get(connectionFactory.cachedConnection)
+ val isClosedValue = isClosedHandle.get(conn)
isClosedValue shouldBe true
}
}
@@ -535,11 +537,12 @@ class JmsConnectorsSpec extends JmsSpec {
completionFuture.failed.futureValue shouldBe a[RuntimeException]
// make sure connection was closed
eventually {
- val isClosedField =
-
connectionFactory.cachedConnection.getClass.getDeclaredFields.filter(x =>
x.getName == "closed").head
- isClosedField.setAccessible(true)
+ val conn = connectionFactory.cachedConnection
+ val connClass = conn.getClass
+ val lookup = MethodHandles.privateLookupIn(connClass,
MethodHandles.lookup())
+ val isClosedHandle = lookup.findVarHandle(connClass, "closed",
classOf[Boolean])
- val isClosedValue =
isClosedField.get(connectionFactory.cachedConnection)
+ val isClosedValue = isClosedHandle.get(conn)
isClosedValue shouldBe true
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
