fix deprecations

tmio · Apr 23, 2024 · 3ba8cb7 · 3ba8cb7
1 parent 07cd889
commit 3ba8cb7
Show file tree

Hide file tree

Showing 25 changed files with 107 additions and 86 deletions.
diff --git a/dependency-versions.gradle b/dependency-versions.gradle
@@ -55,6 +55,7 @@ dependencyManagement {
     dependency('io.opentelemetry:opentelemetry-extension-trace-propagators:1.2.0')
     dependency('io.opentelemetry:opentelemetry-proto:1.7.1-alpha')
     dependency('io.opentelemetry:opentelemetry-sdk:1.36.0')
+    dependency('io.opentelemetry.semconv:opentelemetry-semconv-incubating:1.24.0-alpha')
     dependency('io.opentelemetry:opentelemetry-sdk-trace:1.36.0')
     dependency('io.opentelemetry:opentelemetry-sdk-metrics:1.36.0')
     dependency('io.opentelemetry:opentelemetry-sdk-testing:1.36.0')
@@ -90,7 +91,7 @@ dependencyManagement {
       entry 'bcpkix-jdk15on'
       entry 'bcprov-jdk15on'
     }
-    dependencySet(group: 'org.eclipse.jetty', version: '12.0.7') {
+    dependencySet(group: 'org.eclipse.jetty', version: '11.0.20') {
       entry 'jetty-server'
       entry 'jetty-servlet'
       entry 'jetty-servlets'

diff --git a/devp2p-proxy/src/main/kotlin/org/apache/tuweni/devp2p/proxy/TcpDownstream.kt b/devp2p-proxy/src/main/kotlin/org/apache/tuweni/devp2p/proxy/TcpDownstream.kt
@@ -6,12 +6,11 @@ import io.vertx.core.Vertx
 import io.vertx.core.buffer.Buffer
 import io.vertx.core.net.NetServer
 import io.vertx.core.net.NetSocket
-import io.vertx.kotlin.coroutines.await
+import io.vertx.kotlin.coroutines.coAwait
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
 import org.apache.tuweni.bytes.Bytes
-import org.apache.tuweni.concurrent.coroutines.await
 import org.slf4j.LoggerFactory
 import kotlin.coroutines.CoroutineContext
 
@@ -34,7 +33,7 @@ class TcpDownstream(
     val server = vertx.createNetServer()
     server.connectHandler {
       handleSocket(it)
-    }.listen(port, host).await()
+    }.listen(port, host).coAwait()
     tcpServer = server
     logger.info("Started downstream proxy server on $host:$port")
   }

diff --git a/devp2p-proxy/src/main/kotlin/org/apache/tuweni/devp2p/proxy/TcpUpstream.kt b/devp2p-proxy/src/main/kotlin/org/apache/tuweni/devp2p/proxy/TcpUpstream.kt
@@ -6,6 +6,7 @@ import io.vertx.core.Vertx
 import io.vertx.core.buffer.Buffer
 import io.vertx.core.net.NetClient
 import io.vertx.kotlin.coroutines.await
+import io.vertx.kotlin.coroutines.coAwait
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
@@ -33,7 +34,7 @@ class TcpUpstream(
   }
 
   override suspend fun handleRequest(message: Bytes): Bytes {
-    val socket = tcpclient!!.connect(port, host).await()
+    val socket = tcpclient!!.connect(port, host).coAwait()
 
     val result = AsyncResult.incomplete<Bytes>()
     socket.handler {
@@ -44,7 +45,7 @@ class TcpUpstream(
     }.exceptionHandler {
       result.completeExceptionally(it)
     }
-    socket.write(Buffer.buffer(message.toArrayUnsafe())).await()
+    socket.write(Buffer.buffer(message.toArrayUnsafe())).coAwait()
 
     return result.await()
   }

diff --git a/devp2p/src/main/kotlin/org/apache/tuweni/devp2p/DiscoveryService.kt b/devp2p/src/main/kotlin/org/apache/tuweni/devp2p/DiscoveryService.kt
@@ -9,6 +9,7 @@ import io.vertx.core.buffer.Buffer
 import io.vertx.core.datagram.DatagramPacket
 import io.vertx.core.net.SocketAddress
 import io.vertx.kotlin.coroutines.await
+import io.vertx.kotlin.coroutines.coAwait
 import io.vertx.kotlin.coroutines.dispatcher
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineExceptionHandler
@@ -335,7 +336,7 @@ internal class CoroutineDiscoveryService constructor(
   }
 
   fun start() = launch {
-    server.handler { receiveDatagram(it) }.listen(bindAddress.port(), bindAddress.host()).await()
+    server.handler { receiveDatagram(it) }.listen(bindAddress.port(), bindAddress.host()).coAwait()
     val endpoint = Endpoint(
       advertiseAddress ?: (server.localAddress()).host(),
       advertiseUdpPort ?: server.localAddress().port(),
@@ -428,7 +429,7 @@ internal class CoroutineDiscoveryService constructor(
   override suspend fun shutdown() {
     if (shutdown.compareAndSet(false, true)) {
       logger.info("{}: shutdown", serviceDescriptor)
-      server.close().await()
+      server.close().coAwait()
       for (pending in awaitingPongs.values) {
         pending.complete(null)
       }
@@ -916,6 +917,6 @@ internal class CoroutineDiscoveryService constructor(
   }
 
   private suspend fun sendPacket(address: SocketAddress, packet: Packet) {
-    server.send(Buffer.buffer(packet.encode().toArrayUnsafe()), address.port(), address.host()).await()
+    server.send(Buffer.buffer(packet.encode().toArrayUnsafe()), address.port(), address.host()).coAwait()
   }
 }
diff --git a/devp2p/src/main/kotlin/org/apache/tuweni/devp2p/PacketType.kt b/devp2p/src/main/kotlin/org/apache/tuweni/devp2p/PacketType.kt
@@ -6,6 +6,8 @@ import org.apache.tuweni.bytes.Bytes
 import org.apache.tuweni.bytes.Bytes32
 import org.apache.tuweni.crypto.SECP256K1
 
+private const val MAX_VALUE: Byte = 0x7f
+
 /**
  * DevP2P discovery packet types
  * @param typeId the byte representing the type
@@ -86,7 +88,6 @@ internal enum class PacketType(
   }, ;
 
   companion object {
-    private const val MAX_VALUE: Byte = 0x7f
     private val INDEX = arrayOfNulls<PacketType?>(MAX_VALUE.toInt())
 
     init {
@@ -100,7 +101,7 @@ internal enum class PacketType(
   }
 
   init {
-    require(typeId <= PacketType.MAX_VALUE) { "Packet typeId must be in range [0x00, 0x80)" }
+    require(typeId <= MAX_VALUE) { "Packet typeId must be in range [0x00, 0x80)" }
   }
 
   abstract fun decode(

diff --git a/devp2p/src/main/kotlin/org/apache/tuweni/devp2p/v5/DiscoveryV5Service.kt b/devp2p/src/main/kotlin/org/apache/tuweni/devp2p/v5/DiscoveryV5Service.kt
@@ -7,6 +7,7 @@ import io.vertx.core.buffer.Buffer
 import io.vertx.core.datagram.DatagramPacket
 import io.vertx.core.net.SocketAddress
 import io.vertx.kotlin.coroutines.await
+import io.vertx.kotlin.coroutines.coAwait
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -176,38 +177,38 @@ internal class DefaultDiscoveryV5Service(
   private lateinit var receiveJob: Job
 
   override suspend fun start(): AsyncCompletion {
-    server.handler(this::receiveDatagram).listen(bindAddress.port, bindAddress.hostString).await()
+    server.handler(this::receiveDatagram).listen(bindAddress.port, bindAddress.hostString).coAwait()
     return bootstrap()
   }
 
   override suspend fun terminate() {
     if (started.compareAndSet(true, false)) {
       receiveJob.cancel()
-      server.close().await()
+      server.close().coAwait()
     }
   }
 
   override fun enr(): EthereumNodeRecord = selfEnr
 
   override suspend fun addPeer(enr: EthereumNodeRecord, address: SocketAddress): AsyncCompletion {
     val session = sessions[address]
-    if (session == null) {
+    return if (session == null) {
       logger.trace("Creating new session for peer {}", enr)
       val handshakeSession = handshakes.computeIfAbsent(address) { addr -> createHandshake(addr, enr.publicKey(), enr) }
-      return asyncCompletion {
+      asyncCompletion {
         logger.trace("Handshake connection start {}", enr)
         handshakeSession.connect().await()
         logger.trace("Handshake connection done {}", enr)
       }
     } else {
       logger.trace("Session found for peer {}", enr)
-      return AsyncCompletion.completed()
+      AsyncCompletion.completed()
     }
   }
 
   private fun send(addr: SocketAddress, message: Bytes) {
     launch {
-      server.send(Buffer.buffer(message.toArrayUnsafe()), addr.port(), addr.host()).await()
+      server.send(Buffer.buffer(message.toArrayUnsafe()), addr.port(), addr.host()).coAwait()
     }
   }
 
@@ -227,7 +228,7 @@ internal class DefaultDiscoveryV5Service(
     var session = sessions.get(packet.sender())
     val size = Math.min(Packet.MAX_SIZE, packet.data().length())
     val buffer = ByteBuffer.allocate(size)
-    packet.data().byteBuf.readBytes(buffer)
+    buffer.put(packet.data().bytes)
     buffer.flip()
     val message = Bytes.wrapByteBuffer(buffer)
     if (message.slice(0, 32) == whoAreYouHeader && session != null) {

diff --git a/devp2p/src/test/kotlin/org/apache/tuweni/devp2p/DiscoveryServiceTest.kt b/devp2p/src/test/kotlin/org/apache/tuweni/devp2p/DiscoveryServiceTest.kt
@@ -6,6 +6,7 @@ import io.vertx.core.Vertx
 import io.vertx.core.buffer.Buffer
 import io.vertx.core.net.SocketAddress
 import io.vertx.kotlin.coroutines.await
+import io.vertx.kotlin.coroutines.coAwait
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import org.apache.tuweni.bytes.Bytes
@@ -61,7 +62,7 @@ internal class DiscoveryServiceTest {
     val reference = AsyncResult.incomplete<Buffer>()
     val client = vertx.createDatagramSocket().handler { res ->
       reference.complete(res.data())
-    }.listen(0, "localhost").await()
+    }.listen(0, "localhost").coAwait()
     val clientEndpoint = Endpoint("192.168.1.1", 5678, 7654)
     val ping = PingPacket.create(
       clientKeyPair,
@@ -70,10 +71,10 @@ internal class DiscoveryServiceTest {
       Endpoint(address),
       null,
     )
-    client.send(Buffer.buffer(ping.encode().toArrayUnsafe()), address.port(), address.host()).await()
+    client.send(Buffer.buffer(ping.encode().toArrayUnsafe()), address.port(), address.host()).coAwait()
     val datagram = reference.await()
     val buffer = ByteBuffer.allocate(datagram.length())
-    datagram.byteBuf.readBytes(buffer)
+    buffer.put(datagram.bytes)
     val pong = Packet.decodeFrom(buffer) as PongPacket
     assertEquals(discoveryService.nodeId, pong.nodeId)
     assertEquals(ping.hash, pong.pingHash)
@@ -92,7 +93,7 @@ internal class DiscoveryServiceTest {
     reference.set(AsyncResult.incomplete())
     val bootstrapClient = vertx.createDatagramSocket().handler { res ->
       reference.get().complete(res.data())
-    }.listen(0, "127.0.0.1").await()
+    }.listen(0, "127.0.0.1").coAwait()
 
     val serviceKeyPair = SECP256K1.KeyPair.random()
     val peerRepository = EphemeralPeerRepository()
@@ -113,7 +114,7 @@ internal class DiscoveryServiceTest {
 
     val datagram = reference.get().await()
     val buffer = ByteBuffer.allocate(datagram.length())
-    datagram.byteBuf.readBytes(buffer)
+    buffer.put(datagram.bytes)
     val ping = Packet.decodeFrom(buffer) as PingPacket
     assertEquals(discoveryService.nodeId, ping.nodeId)
     assertEquals(
@@ -132,7 +133,7 @@ internal class DiscoveryServiceTest {
     )
     reference.set(AsyncResult.incomplete())
     val address = SocketAddress.inetSocketAddress(discoveryService.localPort, "127.0.0.1")
-    bootstrapClient.send(Buffer.buffer(pong.encode().toArrayUnsafe()), address.port(), address.host()).await()
+    bootstrapClient.send(Buffer.buffer(pong.encode().toArrayUnsafe()), address.port(), address.host()).coAwait()
 
     val findNodesDatagram = reference.get().await()
 
@@ -164,7 +165,7 @@ internal class DiscoveryServiceTest {
     val reference = AsyncResult.incomplete<Buffer>()
     val bootstrapClient = vertx.createDatagramSocket().handler { res ->
       reference.complete(res.data())
-    }.listen(0, "localhost").await()
+    }.listen(0, "localhost").coAwait()
 
     val serviceKeyPair = SECP256K1.KeyPair.random()
     val peerRepository = EphemeralPeerRepository()
@@ -184,7 +185,7 @@ internal class DiscoveryServiceTest {
     )
     val datagram = reference.await()
     val buffer = ByteBuffer.allocate(datagram.length())
-    datagram.byteBuf.readBytes(buffer)
+    buffer.put(datagram.bytes)
     val ping = Packet.decodeFrom(buffer) as PingPacket
     assertEquals(discoveryService.nodeId, ping.nodeId)
     assertEquals(
@@ -202,7 +203,7 @@ internal class DiscoveryServiceTest {
       null,
     )
     val address = SocketAddress.inetSocketAddress(discoveryService.localPort, "127.0.0.1")
-    bootstrapClient.send(Buffer.buffer(pong.encode().toArrayUnsafe()), address.port(), address.host()).await()
+    bootstrapClient.send(Buffer.buffer(pong.encode().toArrayUnsafe()), address.port(), address.host()).coAwait()
 
     delay(1000)
     val bootstrapPeer =
@@ -225,7 +226,7 @@ internal class DiscoveryServiceTest {
     val reference = AsyncResult.incomplete<Buffer>()
     val bootstrapClient = vertx.createDatagramSocket().handler { res ->
       reference.complete(res.data())
-    }.listen(0, "localhost").await()
+    }.listen(0, "localhost").coAwait()
 
     val discoveryService = DiscoveryService.open(
       vertx,
@@ -244,7 +245,7 @@ internal class DiscoveryServiceTest {
 
     val datagram = reference.await()
     val buffer = ByteBuffer.allocate(datagram.length())
-    datagram.byteBuf.readBytes(buffer)
+    buffer.put(datagram.bytes)
     val ping = Packet.decodeFrom(buffer) as PingPacket
     assertEquals(discoveryService.nodeId, ping.nodeId)
     assertEquals(
@@ -264,7 +265,7 @@ internal class DiscoveryServiceTest {
     reference.set(AsyncResult.incomplete())
     val bootstrapClient = vertx.createDatagramSocket().handler { res ->
       reference.get().complete(res.data())
-    }.listen(0, "localhost").await()
+    }.listen(0, "localhost").coAwait()
 
     val discoveryService = DiscoveryService.open(
       vertx,
@@ -280,7 +281,7 @@ internal class DiscoveryServiceTest {
     val datagram1 = reference.get().await()
     reference.set(AsyncResult.incomplete())
     val buffer1 = ByteBuffer.allocate(datagram1.length())
-    datagram1.byteBuf.readBytes(buffer1)
+    buffer1.put(datagram1.bytes)
     val ping1 = Packet.decodeFrom(buffer1) as PingPacket
     assertEquals(discoveryService.nodeId, ping1.nodeId)
     assertEquals(
@@ -290,7 +291,7 @@ internal class DiscoveryServiceTest {
     val datagram2 = reference.get().await()
     reference.set(AsyncResult.incomplete())
     val buffer2 = ByteBuffer.allocate(datagram2.length())
-    datagram2.byteBuf.readBytes(buffer2)
+    buffer2.put(datagram2.bytes)
     val ping2 = Packet.decodeFrom(buffer2) as PingPacket
     assertEquals(discoveryService.nodeId, ping2.nodeId)
     assertEquals(
@@ -300,7 +301,7 @@ internal class DiscoveryServiceTest {
     val datagram3 = reference.get().await()
     reference.set(AsyncResult.incomplete())
     val buffer3 = ByteBuffer.allocate(datagram3.length())
-    datagram3.byteBuf.readBytes(buffer3)
+    buffer3.put(datagram3.bytes)
     val ping3 = Packet.decodeFrom(buffer3) as PingPacket
     assertEquals(discoveryService.nodeId, ping3.nodeId)
     assertEquals(
@@ -328,18 +329,18 @@ internal class DiscoveryServiceTest {
     reference.set(AsyncResult.incomplete())
     val client = vertx.createDatagramSocket().handler { res ->
       reference.get().complete(res.data())
-    }.listen(0, "localhost").await()
+    }.listen(0, "localhost").coAwait()
     val findNodes =
       FindNodePacket.create(
         clientKeyPair,
         System.currentTimeMillis(),
         SECP256K1.KeyPair.random().publicKey(),
       )
-    client.send(Buffer.buffer(findNodes.encode().toArrayUnsafe()), address.port(), address.host()).await()
+    client.send(Buffer.buffer(findNodes.encode().toArrayUnsafe()), address.port(), address.host()).coAwait()
 
     val datagram = reference.get().await()
     val buffer = ByteBuffer.allocate(datagram.length())
-    datagram.byteBuf.readBytes(buffer)
+    buffer.put(datagram.bytes)
     val ping = Packet.decodeFrom(buffer) as PingPacket
     assertEquals(discoveryService.nodeId, ping.nodeId)
 
@@ -355,11 +356,11 @@ internal class DiscoveryServiceTest {
     )
 
     reference.set(AsyncResult.incomplete())
-    client.send(Buffer.buffer(pong.encode().toArrayUnsafe()), address.port(), address.host()).await()
+    client.send(Buffer.buffer(pong.encode().toArrayUnsafe()), address.port(), address.host()).coAwait()
 
     val datagram2 = reference.get().await()
     val buffer2 = ByteBuffer.allocate(datagram2.length())
-    datagram2.byteBuf.readBytes(buffer2)
+    buffer2.put(datagram2.bytes)
     val neighbors = Packet.decodeFrom(buffer2) as NeighborsPacket
     assertEquals(discoveryService.nodeId, neighbors.nodeId)
 

diff --git a/devp2p/src/test/kotlin/org/apache/tuweni/devp2p/v5/DefaultDiscoveryV5ServiceTest.kt b/devp2p/src/test/kotlin/org/apache/tuweni/devp2p/v5/DefaultDiscoveryV5ServiceTest.kt
@@ -5,6 +5,7 @@ package org.apache.tuweni.devp2p.v5
 import io.vertx.core.Vertx
 import io.vertx.core.buffer.Buffer
 import io.vertx.kotlin.coroutines.await
+import io.vertx.kotlin.coroutines.coAwait
 import kotlinx.coroutines.runBlocking
 import org.apache.tuweni.bytes.Bytes
 import org.apache.tuweni.concurrent.AsyncResult
@@ -52,7 +53,7 @@ class DefaultDiscoveryV5ServiceTest {
     val reference = AsyncResult.incomplete<Buffer>()
     val client = vertx.createDatagramSocket().handler { res ->
       reference.complete(res.data())
-    }.listen(19001, "localhost").await()
+    }.listen(19001, "localhost").coAwait()
     val discoveryV5Service: DiscoveryV5Service =
       DiscoveryService.open(
         vertx,
@@ -64,7 +65,7 @@ class DefaultDiscoveryV5ServiceTest {
 
     val datagram = reference.await()
     val buffer = ByteBuffer.allocate(datagram.length())
-    datagram.byteBuf.readBytes(buffer)
+    buffer.put(datagram.bytes)
     buffer.flip()
     val receivedBytes = Bytes.wrapByteBuffer(buffer)
     val content = receivedBytes.slice(45)