New API, inMemorySocketPair() (#1638)

* New API, inMemorySocketPair()

I'm using Array<Socket> instead of Pair<Socket, Socket>
because Array is _slightly_ more friendly to Java-language
consumers.

* apiDump

---------

Co-authored-by: Jesse Wilson <[email protected]>

Author: swankjesse

okio/api/okio.api

@@ -650,6 +650,7 @@ public final class okio/Okio {
 	public static final fun hashingSink (Lokio/Sink;Ljavax/crypto/Mac;)Lokio/HashingSink;
 	public static final fun hashingSource (Lokio/Source;Ljava/security/MessageDigest;)Lokio/HashingSource;
 	public static final fun hashingSource (Lokio/Source;Ljavax/crypto/Mac;)Lokio/HashingSource;
+	public static final fun inMemorySocketPair (J)[Lokio/Socket;
 	public static final fun openZip (Lokio/FileSystem;Lokio/Path;)Lokio/FileSystem;
 	public static final fun sink (Ljava/io/File;)Lokio/Sink;
 	public static final fun sink (Ljava/io/File;Z)Lokio/Sink;

okio/src/jvmMain/kotlin/okio/JvmOkio.kt

@@ -34,6 +34,7 @@ import java.security.MessageDigest
 import javax.crypto.Cipher
 import javax.crypto.Mac
 import okio.internal.DefaultSocket
+import okio.internal.PipeSocket
 import okio.internal.ResourceFileSystem
 import okio.internal.SocketAsyncTimeout
 import okio.internal.isAndroidGetsocknameError
@@ -142,6 +143,36 @@ fun Socket.source(): Source {
 @JvmName("socket")
 fun Socket.asOkioSocket(): okio.Socket = DefaultSocket(this)
 
+/**
+ * Returns an array of two symmetric sockets, _A_ (element 0) and _B_ (element 1) that are mutually
+ * connected:
+ *
+ *  * Pipe AB connects _A_’s sink to _B_’s source.
+ *  * Pipe BA connects _B_’s sink to _A_’s source.
+ *
+ * Each pipe uses a buffer to decouple source and sink. This buffer has a user-specified maximum
+ * size. When a socket writer outruns its corresponding reader, the buffer fills up and eventually
+ * writes to the sink will block until the reader has caught up. Symmetrically, if a reader outruns
+ * its writer, reads block until there is data to be read.
+ *
+ * There is a buffer for Pipe AB and another for Pipe BA. The maximum amount of memory that could be
+ * held by the two sockets together is `maxBufferSize * 2`.
+ *
+ * Limit the amount of time spent waiting for the other party by configuring [timeouts][Timeout] on
+ * the source and the sink.
+ *
+ * When the sink is closed, source reads will continue to complete normally until the buffer is
+ * exhausted. At that point reads will return -1, indicating the end of the stream. But if the
+ * source is closed first, writes to the sink will immediately fail with an [IOException].
+ *
+ * Canceling either socket immediately fails all reads and writes on both sockets.
+ */
+fun inMemorySocketPair(maxBufferSize: Long): Array<okio.Socket> {
+  val ab = Pipe(maxBufferSize)
+  val ba = Pipe(maxBufferSize)
+  return arrayOf(PipeSocket(ab, ba), PipeSocket(ba, ab))
+}
+
 /** Returns a sink that writes to `file`. */
 @JvmOverloads
 @Throws(FileNotFoundException::class)

okio/src/jvmMain/kotlin/okio/internal/PipeSocket.kt

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2025 Square, Inc.
+ *
+ * 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.
+ */
+package okio.internal
+
+import okio.Pipe
+import okio.Sink
+import okio.Socket
+import okio.Source
+
+internal class PipeSocket(val sinkPipe: Pipe, val sourcePipe: Pipe) : Socket {
+  override val source: Source
+    get() = sourcePipe.source
+
+  override val sink: Sink
+    get() = sinkPipe.sink
+
+  override fun cancel() {
+    sourcePipe.cancel()
+    sinkPipe.cancel()
+  }
+}

okio/src/jvmTest/kotlin/okio/SocketTest.kt

@@ -46,8 +46,8 @@ class SocketTest(val factory: Factory = Factory.Default) {
   @Before
   fun setUp() {
     val socketPair = factory.createSocketPair()
-    this.socket = socketPair.first
-    this.peerSocket = socketPair.second
+    this.socket = socketPair[0]
+    this.peerSocket = socketPair[1]
     this.peer = AsyncSocket(peerSocket)
   }
 
@@ -257,7 +257,7 @@ class SocketTest(val factory: Factory = Factory.Default) {
   enum class Factory {
     /** Implements an okio.Socket using the `java.net.Socket` API on OS sockets. */
     Default {
-      override fun createSocketPair(): Pair<Socket, Socket> {
+      override fun createSocketPair(): Array<Socket> {
         val localhost = InetAddress.getByName("localhost")
 
         val serverSocket = ServerSocket()
@@ -272,33 +272,15 @@ class SocketTest(val factory: Factory = Factory.Default) {
         socketA.connect(InetSocketAddress(localhost, serverSocket.localPort))
 
         val socketB = socketBFuture.get()
-        return socketA.asOkioSocket() to socketB.asOkioSocket()
+        return arrayOf(socketA.asOkioSocket(), socketB.asOkioSocket())
       }
     },
 
-    /** Implements an okio.Socket using a pair of in-memory pipes. */
     Pipes {
-      override fun createSocketPair(): Pair<Socket, Socket> {
-        class PipeSocket(val sourcePipe: Pipe, val sinkPipe: Pipe) : Socket {
-          override val source: Source
-            get() = sourcePipe.source
-          override val sink: Sink
-            get() = sinkPipe.sink
-
-          override fun cancel() {
-            sourcePipe.cancel()
-            sinkPipe.cancel()
-          }
-        }
-
-        val aToB = Pipe(1024)
-        val bToA = Pipe(1024)
-
-        return PipeSocket(aToB, bToA) to PipeSocket(bToA, aToB)
-      }
+      override fun createSocketPair() = inMemorySocketPair(1024)
     };
 
-    /** Returns a pair of mutually-connected sockets. */
-    abstract fun createSocketPair(): Pair<Socket, Socket>
+    /** Returns two mutually-connected sockets. */
+    abstract fun createSocketPair(): Array<Socket>
   }
 }