MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-09-27 10:03:11 +07:00
Code Issues Packages Projects Releases Wiki Activity

[eel] IJPL-184988: Get rid of EelResult

Browse Source 
- remove results from the channels

GitOrigin-RevId: 25c5ab5cfcc56c5bf62e53427e7702de1e3368a3
This commit is contained in:
Andrii Zinchenko
2025-05-15 08:24:55 +00:00
committed by intellij-monorepo-bot
parent 11b88df709
commit 5ace6d13eb
27 changed files with 369 additions and 445 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/eel-impl/src/com/intellij/platform/eel/impl/local/LocalEelPosixProcess.kt
+3 -3
View File
@@ -27,9 +27,9 @@ internal class LocalEelPosixProcess(
ApplicationManager.getApplication().service<EelLocalApiService>().scope("LocalEelProcess pid=${process.pid()}")
override val pid: EelApi.Pid = LocalPid(process.pid())
override val stdin: EelSendChannel<IOException> = process.outputStream.asEelChannel()
override val stdout: EelReceiveChannel<IOException> = StreamClosedAwareEelReceiveChannel(process.inputStream.consumeAsEelChannel())
override val stderr: EelReceiveChannel<IOException> = StreamClosedAwareEelReceiveChannel(process.errorStream.consumeAsEelChannel())
override val stdin: EelSendChannel = process.outputStream.asEelChannel()
override val stdout: EelReceiveChannel = StreamClosedAwareEelReceiveChannel(process.inputStream.consumeAsEelChannel())
override val stderr: EelReceiveChannel = StreamClosedAwareEelReceiveChannel(process.errorStream.consumeAsEelChannel())
override val exitCode: Deferred<Int> = scope.async {
process.awaitExit()
}
platform/eel-impl/src/com/intellij/platform/eel/impl/local/LocalEelWindowsProcess.kt
+3 -4
View File
@@ -27,9 +27,9 @@ internal class LocalEelWindowsProcess(
ApplicationManager.getApplication().service<EelLocalApiService>().scope("LocalEelProcess pid=${process.pid()}")
override val pid: EelApi.Pid = LocalPid(process.pid())
override val stdin: EelSendChannel<IOException> = process.outputStream.asEelChannel()
override val stdout: EelReceiveChannel<IOException> = StreamClosedAwareEelReceiveChannel(process.inputStream.consumeAsEelChannel())
override val stderr: EelReceiveChannel<IOException> = StreamClosedAwareEelReceiveChannel(process.errorStream.consumeAsEelChannel())
override val stdin: EelSendChannel = process.outputStream.asEelChannel()
override val stdout: EelReceiveChannel = StreamClosedAwareEelReceiveChannel(process.inputStream.consumeAsEelChannel())
override val stderr: EelReceiveChannel = StreamClosedAwareEelReceiveChannel(process.errorStream.consumeAsEelChannel())
override val exitCode: Deferred<Int> = scope.async {
process.awaitExit()
}
@@ -52,4 +52,3 @@ internal class LocalEelWindowsProcess(
resizeWindow(WinSize(columns, rows))
}
}
platform/eel-impl/src/com/intellij/platform/eel/impl/local/StreamClosedAwareEelReceiveChannel.kt
+16 -19
View File
@@ -1,10 +1,8 @@
// Copyright 2000-2025 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.impl.local
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult
import com.intellij.platform.eel.channels.EelReceiveChannel
import com.intellij.platform.eel.provider.ResultOkImpl
import java.io.IOException
import java.nio.ByteBuffer
@@ -20,23 +18,22 @@ import java.nio.ByteBuffer
* Even though the data is still lost, there's at least no unexpected error.
*/
internal class StreamClosedAwareEelReceiveChannel(
private val delegate: EelReceiveChannel<IOException>,
) : EelReceiveChannel<IOException> {
override suspend fun receive(dst: ByteBuffer): EelResult<ReadResult, IOException> =
when (val result = delegate.receive(dst)) {
is EelResult.Ok -> result
is EelResult.Error -> {
val e = result.error
if (
e.javaClass == IOException::class.java &&
e.message == "Stream closed" &&
e.stackTrace.firstOrNull()?.className == "java.io.BufferedInputStream"
) {
ResultOkImpl(ReadResult.fromNumberOfReadBytes(-1))
}
else result
}
private val delegate: EelReceiveChannel,
) : EelReceiveChannel {
override suspend fun receive(dst: ByteBuffer): ReadResult {
try {
return delegate.receive(dst)
}
catch (error: IOException) {
val e = error
if (
e.message == "Stream closed" && e.stackTrace.firstOrNull()?.className == "java.io.BufferedInputStream"
) {
return ReadResult.fromNumberOfReadBytes(-1)
}
throw e
}
}
override suspend fun close(): Unit = delegate.close()
}
}
platform/eel-impl/src/com/intellij/platform/eel/impl/local/tunnels/SocketAdapter.kt
+3 -3
View File
@@ -14,8 +14,8 @@ internal class SocketAdapter(channel: SocketChannel) : EelTunnelsApi.Connection
private val socket = channel.socket()
override val sendChannel: EelSendChannel<IOException> = channel.asEelChannel()
override val receiveChannel: EelReceiveChannel<IOException> = channel.consumeAsEelChannel()
override val sendChannel: EelSendChannel = channel.asEelChannel()
override val receiveChannel: EelReceiveChannel = channel.consumeAsEelChannel()
override suspend fun configureSocket(block: suspend ConfigurableClientSocket.() -> Unit) {
block(ConfigurableClientSocketImpl(socket))
@@ -31,4 +31,4 @@ internal class SocketAdapter(channel: SocketChannel) : EelTunnelsApi.Connection
}
socket.close()
}
}
}
platform/eel-impl/src/com/intellij/platform/eel/impl/local/tunnels/localTunnelsImpl.kt
+13 -14
View File
@@ -194,20 +194,19 @@ private class ConnectionAcceptorImpl(private val boundServerSocket: ServerSocket
@Service
private class MyService(val scope: CoroutineScope)
private suspend fun copyWithLoggingAndErrorHandling(src: EelReceiveChannel<IOException>, dest: EelSendChannel<IOException>, title: String, onError: (IOException) -> Unit) {
when (val r = copy(src, dest)) {
is EelResult.Error -> {
when (val e = r.error) {
is CopyResultError.InError -> {
logger.warn("$title input error", e.inError)
onError(e.inError)
}
is CopyResultError.OutError -> {
logger.warn("$title output error", e.outError)
onError(e.outError)
}
private suspend fun copyWithLoggingAndErrorHandling(src: EelReceiveChannel, dest: EelSendChannel, title: String, onError: (IOException) -> Unit) {
try {
copy(src, dest)
} catch (e: CopyError) {
when (e) {
is CopyError.InError -> {
logger.warn("$title input error", e.cause)
onError(e.cause as IOException)
}
is CopyError.OutError -> {
logger.warn("$title output error", e.cause)
onError(e.cause as IOException)
}
}
is EelResult.Ok -> Unit
}
}
}
platform/eel-provider/src/com/intellij/platform/eel/provider/ResultImpl.kt
-7
View File
@@ -1,7 +0,0 @@
// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.provider
import com.intellij.platform.eel.EelResult
data class ResultOkImpl<P>(override val value: P) : EelResult.Ok<P>
data class ResultErrImpl<E>(override val error: E) : EelResult.Error<E>
platform/eel-provider/src/com/intellij/platform/eel/provider/utils/eelChannelTextUtils.kt
+5 -10
View File
@@ -1,13 +1,10 @@
// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.provider.utils
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult.*
import com.intellij.platform.eel.channels.EelReceiveChannel
import com.intellij.platform.eel.channels.EelSendChannel
import com.intellij.platform.eel.channels.sendWholeBuffer
import com.intellij.platform.eel.getOr
import com.intellij.platform.eel.provider.ResultOkImpl
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import org.jetbrains.annotations.CheckReturnValue
@@ -22,17 +19,15 @@ private val DEFAULT_CHARSET = Charsets.UTF_8
/**
* Send [text] till the end. Useful to send commands to process
*/
@CheckReturnValue
suspend fun <ERR : Any> EelSendChannel<ERR>.sendWholeText(text: String, charset: Charset = DEFAULT_CHARSET): EelResult<Unit, ERR> = sendWholeBuffer(ByteBuffer.wrap(text.toByteArray(charset)))
suspend fun EelSendChannel.sendWholeText(text: String, charset: Charset = DEFAULT_CHARSET): Unit = sendWholeBuffer(ByteBuffer.wrap(text.toByteArray(charset)))
/**
* Reads text from the channel to the end.
*/
suspend fun <ERR : Any> EelReceiveChannel<ERR>.readWholeText(bufferSize: Int = DEFAULT_BUFFER_SIZE, charset: Charset = DEFAULT_CHARSET): EelResult<String, ERR> = withContext(Dispatchers.IO) {
suspend fun EelReceiveChannel.readWholeText(bufferSize: Int = DEFAULT_BUFFER_SIZE, charset: Charset = DEFAULT_CHARSET): String = withContext(Dispatchers.IO) {
val buffer = ByteBuffer.allocate(bufferSize)
val result = ByteArrayOutputStream()
while (receive(buffer).getOr { return@withContext it } != EOF) {
while (receive(buffer) != EOF) {
buffer.flip()
// Redundant copy isn't optimal but ok for text
val tmpBuffer = ByteArray(buffer.limit())
@@ -40,5 +35,5 @@ suspend fun <ERR : Any> EelReceiveChannel<ERR>.readWholeText(bufferSize: Int = D
result.writeBytes(tmpBuffer)
buffer.clear()
}
return@withContext ResultOkImpl(result.toString(charset))
}
return@withContext result.toString(charset)
}
platform/eel-provider/src/com/intellij/platform/eel/provider/utils/eelChannelUtils.kt
+71 -83
View File
@@ -2,13 +2,10 @@
package com.intellij.platform.eel.provider.utils
import com.intellij.openapi.progress.Cancellation.ensureActive
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult.EOF
import com.intellij.platform.eel.channels.EelReceiveChannel
import com.intellij.platform.eel.channels.EelSendChannel
import com.intellij.platform.eel.channels.sendWholeBuffer
import com.intellij.platform.eel.provider.ResultErrImpl
import com.intellij.platform.eel.provider.ResultOkImpl
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
@@ -16,7 +13,6 @@ import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.withContext
import org.jetbrains.annotations.CheckReturnValue
import java.io.IOException
import java.io.InputStream
import java.io.OutputStream
@@ -40,16 +36,16 @@ import kotlin.time.Duration.Companion.milliseconds
// copy functions copy data from/to an eel channel
fun ReadableByteChannel.consumeAsEelChannel(): EelReceiveChannel<IOException> = NioReadToEelAdapter(this)
fun WritableByteChannel.asEelChannel(): EelSendChannel<IOException> = NioWriteToEelAdapter(this)
fun ReadableByteChannel.consumeAsEelChannel(): EelReceiveChannel = NioReadToEelAdapter(this)
fun WritableByteChannel.asEelChannel(): EelSendChannel = NioWriteToEelAdapter(this)
// Flushes data after each writing.
fun OutputStream.asEelChannel(): EelSendChannel<IOException> = NioWriteToEelAdapter(Channels.newChannel(this), this)
fun InputStream.consumeAsEelChannel(): EelReceiveChannel<IOException> = NioReadToEelAdapter(Channels.newChannel(this))
fun EelReceiveChannel<IOException>.consumeAsInputStream(blockingContext: CoroutineContext = Dispatchers.IO): InputStream =
fun OutputStream.asEelChannel(): EelSendChannel = NioWriteToEelAdapter(Channels.newChannel(this), this)
fun InputStream.consumeAsEelChannel(): EelReceiveChannel = NioReadToEelAdapter(Channels.newChannel(this))
fun EelReceiveChannel.consumeAsInputStream(blockingContext: CoroutineContext = Dispatchers.IO): InputStream =
InputStreamAdapterImpl(this, blockingContext)
fun EelSendChannel<IOException>.asOutputStream(blockingContext: CoroutineContext = Dispatchers.IO): OutputStream =
fun EelSendChannel.asOutputStream(blockingContext: CoroutineContext = Dispatchers.IO): OutputStream =
OutputStreamAdapterImpl(this, blockingContext)
/**
@@ -57,7 +53,7 @@ fun EelSendChannel<IOException>.asOutputStream(blockingContext: CoroutineContext
* Each buffer is fresh (not reused) but not flipped.
* Errors are thrown out of the channel (directly or wrapped with [IOException] if not throwable).
*/
fun CoroutineScope.consumeReceiveChannelAsKotlin(receiveChannel: EelReceiveChannel<*>, bufferSize: Int = DEFAULT_BUFFER_SIZE): ReceiveChannel<ByteBuffer> =
fun CoroutineScope.consumeReceiveChannelAsKotlin(receiveChannel: EelReceiveChannel, bufferSize: Int = DEFAULT_BUFFER_SIZE): ReceiveChannel<ByteBuffer> =
consumeReceiveChannelAsKotlinImpl(receiveChannel, bufferSize)
/**
@@ -69,23 +65,22 @@ fun CoroutineScope.consumeReceiveChannelAsKotlin(receiveChannel: EelReceiveChann
* As soon as channel gets closed -- flow finishes.
* ```kotlin
* suspend fun chat(process: EelProcess) {
* process.stdout.lines().collect {
* val line = it.getOrThrow()
* process.stdout.lines().collect { line ->
* if (line == "Nice to meet you") {
* process.stdin.sendWholeText("Nice to meet you too!").getOrThrow()
* process.stdin.sendWholeText("Nice to meet you too!")
* }
* if (line == "Who are you?") {
* process.stdin.sendWholeText("An eel api!").getOrThrow()
* process.stdin.sendWholeText("An eel api!")
* }
* }
* }
* ```
*/
fun <E : Any> EelReceiveChannel<E>.lines(charset: Charset): Flow<EelResult<String, E>> = linesImpl(charset)
fun <E : Any> EelReceiveChannel<E>.lines(): Flow<EelResult<String, E>> = lines(Charset.defaultCharset())
fun EelReceiveChannel.lines(charset: Charset): Flow<String> = linesImpl(charset)
fun EelReceiveChannel.lines(): Flow<String> = lines(Charset.defaultCharset())
fun Socket.consumeAsEelChannel(): EelReceiveChannel<IOException> = consumeAsEelChannelImpl()
fun Socket.asEelChannel(): EelSendChannel<IOException> = asEelChannelImpl()
fun Socket.consumeAsEelChannel(): EelReceiveChannel = consumeAsEelChannelImpl()
fun Socket.asEelChannel(): EelSendChannel = asEelChannelImpl()
/**
* Bidirectional [kotlinx.coroutines.channels.Channel.RENDEZVOUS] pipe much like [java.nio.channels.Pipe].
@@ -94,8 +89,8 @@ fun Socket.asEelChannel(): EelSendChannel<IOException> = asEelChannelImpl()
* Calling [closePipe] closes both [sink] and [source], you might provide custom error that will be reported on a writing attempt.
*/
interface EelPipe {
val sink: EelSendChannel<IOException>
val source: EelReceiveChannel<IOException>
val sink: EelSendChannel
val source: EelReceiveChannel
fun closePipe(error: Throwable? = null)
}
@@ -106,22 +101,17 @@ fun EelPipe(): EelPipe = EelPipeImpl()
* Reads all data till the end from a channel.
* Semantics is the same as [InputStream.readAllBytes]
*/
suspend fun EelReceiveChannel<IOException>.readAllBytes(): EelResult<ByteArray, IOException> = withContext(Dispatchers.IO) {
suspend fun EelReceiveChannel.readAllBytes(): ByteArray = withContext(Dispatchers.IO) {
// The current implementation is suboptimal and might be rewritten, but the API should be the same
try {
ResultOkImpl(consumeAsInputStream().readAllBytes())
}
catch (e: IOException) {
ResultErrImpl(e)
}
consumeAsInputStream().readAllBytes()
}
/**
* Result of [copy]
*/
sealed interface CopyResultError<ERR_IN : Any, ERR_OUT : Any> {
data class InError<ERR_IN : Any, ERR_OUT : Any>(val inError: ERR_IN) : CopyResultError<ERR_IN, ERR_OUT>
data class OutError<ERR_IN : Any, ERR_OUT : Any>(val outError: ERR_OUT) : CopyResultError<ERR_IN, ERR_OUT>
sealed class CopyError(override val cause: Throwable) : IOException() {
class InError(override val cause: Throwable) : CopyError(cause)
class OutError(override val cause: Throwable) : CopyError(cause)
}
@@ -139,69 +129,67 @@ enum class OnError {
/**
* Copies from [src] to [dst] till [src]'s end ([com.intellij.platform.eel.ReadResult.EOF]).
* Result is either OK or [CopyResultError.InError] (when src returned an error) or [CopyResultError.OutError] (when dst did so)
* The function completes successfully or throws [CopyError.InError] (when src returned an error) or [CopyError.OutError] (when dst did so)
* Channels aren't closed.
* [onWriteError] and [onReadError] might be used to configure read/write error processing
*/
@CheckReturnValue
suspend fun <ERR_IN : Any, ERR_OUT : Any> copy(
src: EelReceiveChannel<ERR_IN>,
dst: EelSendChannel<ERR_OUT>,
@Throws(CopyError::class)
suspend fun copy(
src: EelReceiveChannel,
dst: EelSendChannel,
bufferSize: Int = DEFAULT_BUFFER_SIZE,
// CPU-bound, but due to the mokk error can't be used in tests. Used default in prod.
dispatcher: CoroutineDispatcher = Dispatchers.Default,
onReadError: suspend (ERR_IN) -> OnError = { OnError.EXIT },
onWriteError: suspend (ERR_OUT) -> OnError = { OnError.EXIT },
): EelResult<Unit, CopyResultError<ERR_IN, ERR_OUT>> =
withContext(dispatcher) {
assert(bufferSize > 0)
var sendBackoff = backoff()
var receiveBackoff = backoff()
val buffer = ByteBuffer.allocate(bufferSize)
while (true) {
buffer.clear()
// read data
when (val r = src.receive(buffer)) {
is EelResult.Error -> {
when (onReadError(r.error)) {
OnError.RETRY -> {
delay(receiveBackoff.next())
continue
}
OnError.EXIT -> return@withContext ResultErrImpl(CopyResultError.InError(r.error))
}
}
is EelResult.Ok -> if (r.value == EOF) {
break
}
else {
receiveBackoff = backoff()
}
}
buffer.flip()
do {
// write data
when (val r = dst.sendWholeBuffer(buffer)) {
is EelResult.Error -> {
when (onWriteError(r.error)) {
OnError.RETRY -> {
delay(sendBackoff.next())
continue
}
OnError.EXIT -> return@withContext ResultErrImpl(CopyResultError.OutError(r.error))
}
onReadError: suspend (IOException) -> OnError = { OnError.EXIT },
onWriteError: suspend (IOException) -> OnError = { OnError.EXIT },
): Unit = withContext(dispatcher) {
assert(bufferSize > 0)
var sendBackoff = backoff()
var receiveBackoff = backoff()
val buffer = ByteBuffer.allocate(bufferSize)
while (true) {
buffer.clear()
// read data
try {
val r = src.receive(buffer)
if (r == EOF) {
break
}
else {
receiveBackoff = backoff()
}
}
catch (error: IOException) {
when (onReadError(error)) {
OnError.RETRY -> {
delay(receiveBackoff.next())
continue
}
OnError.EXIT -> throw CopyError.InError(error)
}
}
buffer.flip()
do {
// write data
try {
dst.sendWholeBuffer(buffer)
sendBackoff = backoff()
}
catch (error: IOException) {
when (onWriteError(error)) {
OnError.RETRY -> {
delay(sendBackoff.next())
continue
}
is EelResult.Ok -> {
sendBackoff = backoff()
}
OnError.EXIT -> throw CopyError.OutError(error)
}
}
while (buffer.hasRemaining())
ensureActive()
}
return@withContext OK_UNIT
while (buffer.hasRemaining())
ensureActive()
}
}
// Slowly increase timeout
private fun backoff(): Iterator<Duration> = ((200..1000 step 200).asSequence() + generateSequence(1000) { 1000 }).map { it.milliseconds }.iterator()
private fun backoff(): Iterator<Duration> = ((200..1000 step 200).asSequence() + generateSequence(1000) { 1000 }).map { it.milliseconds }.iterator()
platform/eel-provider/src/com/intellij/platform/eel/provider/utils/eelChannelUtilsImpl.kt
+47 -80
View File
@@ -1,13 +1,10 @@
// Copyright 2000-2025 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.provider.utils
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult
import com.intellij.platform.eel.channels.EelReceiveChannel
import com.intellij.platform.eel.channels.EelSendChannel
import com.intellij.platform.eel.channels.sendWholeBuffer
import com.intellij.platform.eel.provider.ResultErrImpl
import com.intellij.platform.eel.provider.ResultOkImpl
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
@@ -21,15 +18,10 @@ import java.nio.channels.WritableByteChannel
import java.nio.charset.Charset
import kotlin.coroutines.CoroutineContext
internal class NioReadToEelAdapter(private val readableByteChannel: ReadableByteChannel) : EelReceiveChannel<IOException> {
override suspend fun receive(dst: ByteBuffer): EelResult<ReadResult, IOException> = withContext(Dispatchers.IO) {
return@withContext try {
val read = readableByteChannel.read(dst)
ResultOkImpl(ReadResult.fromNumberOfReadBytes(read))
}
catch (e: IOException) {
ResultErrImpl(e)
}
internal class NioReadToEelAdapter(private val readableByteChannel: ReadableByteChannel) : EelReceiveChannel {
override suspend fun receive(dst: ByteBuffer): ReadResult = withContext(Dispatchers.IO) {
val read = readableByteChannel.read(dst)
ReadResult.fromNumberOfReadBytes(read)
}
override suspend fun close() {
@@ -42,21 +34,14 @@ internal class NioReadToEelAdapter(private val readableByteChannel: ReadableByte
internal class NioWriteToEelAdapter(
private val writableByteChannel: WritableByteChannel,
private val flushable: Flushable? = null,
) : EelSendChannel<IOException> {
) : EelSendChannel {
override val closed: Boolean get() = !writableByteChannel.isOpen
override suspend fun send(src: ByteBuffer): EelResult<Unit, IOException> =
withContext(Dispatchers.IO) {
return@withContext try {
writableByteChannel.write(src)
flushable?.flush()
OK_UNIT
}
catch (e: IOException) {
ResultErrImpl(e)
}
}
override suspend fun send(src: ByteBuffer): Unit = withContext(Dispatchers.IO) {
writableByteChannel.write(src)
flushable?.flush()
}
override suspend fun close() {
withContext(Dispatchers.IO + NonCancellable) {
@@ -71,12 +56,8 @@ internal class NioWriteToEelAdapter(
}
}
val OK_UNIT: ResultOkImpl<Unit> = ResultOkImpl(Unit)
val ERR_CHANNEL_CLOSED: ResultErrImpl<IOException> = ResultErrImpl(IOException("Channel is closed"))
internal class InputStreamAdapterImpl(
private val receiveChannel: EelReceiveChannel<IOException>,
private val receiveChannel: EelReceiveChannel,
private val blockingContext: CoroutineContext,
) : InputStream() {
@@ -112,16 +93,13 @@ internal class InputStreamAdapterImpl(
receiveChannel.receive(dst)
}
when (r) {
is EelResult.Error -> throw IOException(r.error)
is EelResult.Ok -> when (r.value) {
ReadResult.EOF -> {
return -1
}
ReadResult.NOT_EOF -> {
val bytesRead = len - dst.remaining()
if (bytesRead > 0) { // See comment on while(true)
return bytesRead
}
ReadResult.EOF -> {
return -1
}
ReadResult.NOT_EOF -> {
val bytesRead = len - dst.remaining()
if (bytesRead > 0) { // See comment on while(true)
return bytesRead
}
}
}
@@ -130,7 +108,7 @@ internal class InputStreamAdapterImpl(
}
internal class OutputStreamAdapterImpl(
private val sendChannel: EelSendChannel<IOException>,
private val sendChannel: EelSendChannel,
private val blockingContext: CoroutineContext,
) : OutputStream() {
private val oneByte = ByteBuffer.allocate(1)
@@ -147,13 +125,9 @@ internal class OutputStreamAdapterImpl(
@Throws(IOException::class)
private fun write(buffer: ByteBuffer) {
val result = runBlocking(blockingContext) {
runBlocking(blockingContext) {
sendChannel.sendWholeBuffer(buffer)
}
when (result) {
is EelResult.Ok -> Unit
is EelResult.Error -> throw IOException(result.error)
}
}
override fun close() {
@@ -163,64 +137,57 @@ internal class OutputStreamAdapterImpl(
}
}
internal fun CoroutineScope.consumeReceiveChannelAsKotlinImpl(receiveChannel: EelReceiveChannel<*>, bufferSize: Int): ReceiveChannel<ByteBuffer> {
internal fun CoroutineScope.consumeReceiveChannelAsKotlinImpl(receiveChannel: EelReceiveChannel, bufferSize: Int): ReceiveChannel<ByteBuffer> {
val channel = Channel<ByteBuffer>()
launch {
while (true) {
val buffer = ByteBuffer.allocate(bufferSize)
when (val r = receiveChannel.receive(buffer)) {
is EelResult.Error -> {
val cause = r.error
channel.close(cause as? Throwable ?: IOException(cause.toString()))
break
}
is EelResult.Ok -> {
when (r.value) {
ReadResult.EOF -> {
channel.close()
break
}
ReadResult.NOT_EOF -> {
channel.send(buffer.flip())
}
try {
val r = receiveChannel.receive(buffer)
when (r) {
ReadResult.EOF -> {
channel.close()
break
}
ReadResult.NOT_EOF -> {
channel.send(buffer.flip())
}
}
} catch (e: IOException) {
channel.close(e)
break
}
}
}
return channel
}
internal fun Socket.consumeAsEelChannelImpl(): EelReceiveChannel<IOException> =
internal fun Socket.consumeAsEelChannelImpl(): EelReceiveChannel =
channel?.consumeAsEelChannel() ?: inputStream.consumeAsEelChannel()
internal fun Socket.asEelChannelImpl(): EelSendChannel<IOException> =
internal fun Socket.asEelChannelImpl(): EelSendChannel =
channel?.asEelChannel() ?: outputStream.asEelChannel()
internal fun <E : Any> EelReceiveChannel<E>.linesImpl(charset: Charset): Flow<EelResult<String, E>> = flow {
internal fun EelReceiveChannel.linesImpl(charset: Charset): Flow<String> = flow {
val tmpBuffer = ByteBuffer.allocate(1)
var result = ByteArrayOutputStream()
while (true) {
tmpBuffer.rewind()
suspend fun emitBuffer() {
emit(ResultOkImpl(charset.decode(ByteBuffer.wrap(result.toByteArray())).toString()))
emit(charset.decode(ByteBuffer.wrap(result.toByteArray())).toString())
}
when (val r = receive(tmpBuffer)) {
is EelResult.Error -> emit(ResultErrImpl(r.error))
is EelResult.Ok -> {
if (r.value == ReadResult.EOF) {
emitBuffer()
return@flow
}
val b = tmpBuffer.flip().get().toInt()
result.write(b)
if (b == 10) {
emitBuffer()
result = ByteArrayOutputStream()
}
}
val r = receive(tmpBuffer)
if (r == ReadResult.EOF) {
emitBuffer()
return@flow
}
val b = tmpBuffer.flip().get().toInt()
result.write(b)
if (b == 10) {
emitBuffer()
result = ByteArrayOutputStream()
}
}
}
platform/eel-provider/src/com/intellij/platform/eel/provider/utils/eelPipeImpl.kt
+17 -23
View File
@@ -1,14 +1,11 @@
// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.provider.utils
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult
import com.intellij.platform.eel.channels.EelReceiveChannel
import com.intellij.platform.eel.channels.EelSendApi
import com.intellij.platform.eel.channels.EelSendChannel
import com.intellij.platform.eel.channels.EelSendChannelCustomSendWholeBuffer
import com.intellij.platform.eel.getOr
import com.intellij.platform.eel.provider.ResultErrImpl
import com.intellij.platform.eel.provider.ResultOkImpl
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ClosedReceiveChannelException
@@ -19,10 +16,10 @@ import java.nio.ByteBuffer
import java.util.concurrent.ConcurrentLinkedDeque
import java.util.concurrent.atomic.AtomicInteger
internal class EelPipeImpl() : EelPipe, EelReceiveChannel<IOException>, EelSendChannelCustomSendWholeBuffer<IOException> {
internal class EelPipeImpl() : EelPipe, EelReceiveChannel, EelSendChannelCustomSendWholeBuffer {
private companion object {
val OK_EOF = ResultOkImpl(ReadResult.EOF)
val OK_NOT_EOF = ResultOkImpl(ReadResult.NOT_EOF)
val OK_EOF = ReadResult.EOF
val OK_NOT_EOF = ReadResult.NOT_EOF
}
@Volatile
@@ -39,37 +36,37 @@ internal class EelPipeImpl() : EelPipe, EelReceiveChannel<IOException>, EelSendC
private val _bytesInQueue = AtomicInteger(0)
internal val bytesInQueue: Int get() = if (channel.isClosedForSend) 0 else _bytesInQueue.get()
override val sink: EelSendChannel<IOException> = this
override val source: EelReceiveChannel<IOException> = this
override val sink: EelSendChannel = this
override val source: EelReceiveChannel = this
private val sendLocks = ConcurrentLinkedDeque<CompletableDeferred<Unit>>()
override suspend fun sendWholeBufferCustom(src: ByteBuffer): EelResult<Unit, IOException> {
override suspend fun sendWholeBufferCustom(src: ByteBuffer) {
_bytesInQueue.addAndGet(src.remaining())
while (src.hasRemaining()) {
send(src, true).getOr { return it }
send(src, true)
}
return OK_UNIT
}
override suspend fun send(src: ByteBuffer): EelResult<Unit, IOException> {
@EelSendApi
override suspend fun send(src: ByteBuffer) {
return send(src, false)
}
private suspend fun send(src: ByteBuffer, decreaseQueueAfterReceive: Boolean): EelResult<Unit, IOException> {
private suspend fun send(src: ByteBuffer, decreaseQueueAfterReceive: Boolean) {
val sendLock = CompletableDeferred<Unit>()
// `send` should return when buffer is read
sendLocks.add(sendLock)
try {
channel.send(Triple(src, sendLock, decreaseQueueAfterReceive))
return OK_UNIT
return
}
catch (_: ClosedSendChannelException) {
closePipe()
return ERR_CHANNEL_CLOSED
throw IOException("Channel is closed")
}
catch (e: PipeBrokenException) {
closePipe()
return e.asErrorResult()
throw e
}
finally {
sendLock.await() //wait for buffer to read
@@ -77,7 +74,7 @@ internal class EelPipeImpl() : EelPipe, EelReceiveChannel<IOException>, EelSendC
}
}
override suspend fun receive(dst: ByteBuffer): EelResult<ReadResult, IOException> {
override suspend fun receive(dst: ByteBuffer): ReadResult {
val (src, sendLock, decreaseQueueAfterReceive) = try {
channel.receive()
}
@@ -87,7 +84,7 @@ internal class EelPipeImpl() : EelPipe, EelReceiveChannel<IOException>, EelSendC
}
catch (e: PipeBrokenException) {
closePipe()
return e.asErrorResult()
throw e
}
val bytesBeforeRead = src.remaining()
// Choose the best approach:
@@ -135,7 +132,4 @@ internal class EelPipeImpl() : EelPipe, EelReceiveChannel<IOException>, EelSendC
}
}
private class PipeBrokenException(cause: Throwable) : IOException("Pipe was broken with message: ${cause.message}", cause) {
fun asErrorResult(): EelResult.Error<IOException> = ResultErrImpl(this)
}
private class PipeBrokenException(cause: Throwable) : IOException("Pipe was broken with message: ${cause.message}", cause)
platform/eel-provider/src/com/intellij/platform/eel/provider/utils/eelProcessUtils.kt
+2 -4
View File
@@ -5,8 +5,6 @@ package com.intellij.platform.eel.provider.utils
import com.intellij.openapi.util.IntellijInternalApi
import com.intellij.platform.eel.*
import com.intellij.platform.eel.provider.ResultErrImpl
import com.intellij.platform.eel.provider.ResultOkImpl
import com.intellij.platform.eel.provider.getEelDescriptor
import com.intellij.util.io.computeDetached
import kotlinx.coroutines.DelicateCoroutinesApi
@@ -56,11 +54,11 @@ suspend fun EelProcess.awaitProcessResult(): EelProcessExecutionResult {
ByteArrayOutputStream().use { err ->
coroutineScope {
launch {
copy(stdout, out.asEelChannel()).getOrThrow() // TODO: process errors
copy(stdout, out.asEelChannel()) // TODO: process errors
}
launch {
copy(stderr, err.asEelChannel()).getOrThrow() // TODO: process errors
copy(stderr, err.asEelChannel()) // TODO: process errors
}
}
platform/eel-provider/src/com/intellij/platform/eel/provider/utils/eelTunnelUtils.kt
+52 -56
View File
@@ -57,13 +57,13 @@ fun CoroutineScope.forwardLocalPort(tunnels: EelTunnelsApi, localPort: Int, addr
launch {
socket.use {
tunnels.getConnectionToRemotePort().hostAddress(address).withConnectionToRemotePort({
if (it is EelConnectionError.ConnectionProblem) {
LOG.debug("Failed to establish connection $connectionCounter ($localPort - $address: $it); closing socket")
}
else {
LOG.error("Failed to connect to remote port $localPort - $address: $it")
}
}) { (channelTo, channelFrom) ->
if (it is EelConnectionError.ConnectionProblem) {
LOG.debug("Failed to establish connection $connectionCounter ($localPort - $address: $it); closing socket")
}
else {
LOG.error("Failed to connect to remote port $localPort - $address: $it")
}
}) { (channelTo, channelFrom) ->
redirectClientConnectionDataToIJent(currentConnection, socket, channelTo)
redirectIJentDataToClientConnection(currentConnection, socket, channelFrom)
}
@@ -98,9 +98,9 @@ fun CoroutineScope.forwardLocalServer(tunnels: EelTunnelsApi, localPort: Int, ad
val remoteAddress = CompletableDeferred<EelTunnelsApi.ResolvedSocketAddress>()
launch(blockingDispatcher.plus(CoroutineName("Local server on port $localPort (tunneling to $address)"))) {
tunnels.getAcceptorForRemotePort().hostAddress(address).withAcceptorForRemotePort({
LOG.error("Failed to start a server on $address (was forwarding $localPort): $it")
remoteAddress.cancel()
}) { acceptor ->
LOG.error("Failed to start a server on $address (was forwarding $localPort): $it")
remoteAddress.cancel()
}) { acceptor ->
remoteAddress.complete(acceptor.boundAddress)
var connections = 0
for ((sendChannel, receiveChannel) in acceptor.incomingConnections) {
@@ -125,66 +125,62 @@ fun CoroutineScope.forwardLocalServer(tunnels: EelTunnelsApi, localPort: Int, ad
}
@OptIn(DelicateCoroutinesApi::class)
private fun CoroutineScope.redirectClientConnectionDataToIJent(connectionId: Int, socket: Socket, channelToIJent: EelSendChannel<IOException>) = launch(CoroutineName("Reader for connection $connectionId")) {
private fun CoroutineScope.redirectClientConnectionDataToIJent(connectionId: Int, socket: Socket, channelToIJent: EelSendChannel) = launch(CoroutineName("Reader for connection $connectionId")) {
val result = copy(
socket.consumeAsEelChannel(), channelToIJent,
onReadError = {
if (it is SocketTimeoutException && channelToIJent.closed) {
this@launch.cancel("Channel is closed normally")
OnError.EXIT
}
else {
OnError.RETRY
}
},
)
channelToIJent.close()
when (result) {
is EelResult.Ok -> Unit
is EelResult.Error -> {
when (val error = result.error) {
is CopyResultError.InError -> {
if (!socket.isClosed) {
LOG.warn("Connection $connectionId closed", error.inError)
throw CancellationException("Closed because of a socket exception", error.inError)
}
try {
copy(
socket.consumeAsEelChannel(), channelToIJent,
onReadError = {
if (it is SocketTimeoutException && channelToIJent.closed) {
this@launch.cancel("Channel is closed normally")
OnError.EXIT
}
is CopyResultError.OutError -> {
LOG.warn("Connection $connectionId closed by IJent; ${error.outError}")
else {
OnError.RETRY
}
},
)
}
catch (error: IOException) {
when (val error = error) {
is CopyError.InError -> {
if (!socket.isClosed) {
LOG.warn("Connection $connectionId closed", error.cause)
throw CancellationException("Closed because of a socket exception", error.cause)
}
}
is CopyError.OutError -> {
LOG.warn("Connection $connectionId closed by IJent; ${error.cause}")
}
}
}
channelToIJent.close()
}
private fun CoroutineScope.redirectIJentDataToClientConnection(connectionId: Int, socket: Socket, backChannel: EelReceiveChannel<IOException>) = launch(CoroutineName("Writer for connection $connectionId")) {
private fun CoroutineScope.redirectIJentDataToClientConnection(connectionId: Int, socket: Socket, backChannel: EelReceiveChannel) = launch(CoroutineName("Writer for connection $connectionId")) {
val result = copy(backChannel, socket.asEelChannel(),
onReadError = {
if (it is SocketTimeoutException) OnError.RETRY else OnError.EXIT
})
backChannel.close()
when (result) {
is EelResult.Ok -> Unit
is EelResult.Error -> {
when (val error = result.error) {
is CopyResultError.InError -> {
LOG.warn("Error reading from channel", error.inError)
try {
copy(backChannel, socket.asEelChannel(), onReadError = {
if (it is SocketTimeoutException) OnError.RETRY else OnError.EXIT
})
}
catch (error: IOException) {
when (val error = error) {
is CopyError.InError -> {
LOG.warn("Error reading from channel", error.cause)
}
is CopyError.OutError -> {
val e = error.cause
if (!socket.isClosed) {
LOG.warn("Socket connection $connectionId closed", e)
}
is CopyResultError.OutError -> {
val e = error.outError
if (!socket.isClosed) {
LOG.warn("Socket connection $connectionId closed", e)
}
else {
LOG.debug("Socket connection $connectionId closed because of cancellation", e)
}
else {
LOG.debug("Socket connection $connectionId closed because of cancellation", e)
}
}
}
}
backChannel.close()
}
fun EelTunnelsApi.ResolvedSocketAddress.asInetAddress(): InetSocketAddress {
platform/eel/src/com/intellij/platform/eel/EelProcess.kt
+3 -3
View File
@@ -19,9 +19,9 @@ sealed interface EelProcess {
*
* Note that each chunk of data is individually and immediately flushed into the process without any intermediate buffer storage.
*/
val stdin: EelSendChannel<IOException>
val stdout: EelReceiveChannel<IOException>
val stderr: EelReceiveChannel<IOException>
val stdin: EelSendChannel
val stdout: EelReceiveChannel
val stderr: EelReceiveChannel
val exitCode: Deferred<Int>
/**
platform/eel/src/com/intellij/platform/eel/EelTunnelsApi.kt
+6 -6
View File
@@ -48,8 +48,8 @@ sealed interface EelTunnelsApi {
data class ListenOnUnixSocketResult(
val unixSocketPath: String,
val tx: EelSendChannel<IOException>,
val rx: EelReceiveChannel<IOException>,
val tx: EelSendChannel,
val rx: EelReceiveChannel,
)
sealed interface CreateFilePath {
@@ -191,12 +191,12 @@ sealed interface EelTunnelsApi {
/**
* A channel to the server
*/
val sendChannel: EelSendChannel<IOException>
val sendChannel: EelSendChannel
/**
* A channel from the server
*/
val receiveChannel: EelReceiveChannel<IOException>
val receiveChannel: EelReceiveChannel
/**
* Configure various socket options
@@ -295,13 +295,13 @@ interface ConfigurableClientSocket : ConfigurableSocket {
* Convenience operator to decompose connection to a pair of channels when needed.
* @return channel to server
*/
operator fun Connection.component1(): EelSendChannel<IOException> = sendChannel
operator fun Connection.component1(): EelSendChannel = sendChannel
/**
* Convenience operator to decompose connection to a pair of channels when needed.
* @return channel from server
*/
operator fun Connection.component2(): EelReceiveChannel<IOException> = receiveChannel
operator fun Connection.component2(): EelReceiveChannel = receiveChannel
interface EelTunnelsPosixApi : EelTunnelsApi {
platform/eel/src/com/intellij/platform/eel/channels/EelReceiveChannel.kt
+3 -5
View File
@@ -1,7 +1,6 @@
// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.channels
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult
import org.jetbrains.annotations.CheckReturnValue
import java.nio.ByteBuffer
@@ -9,15 +8,14 @@ import java.nio.ByteBuffer
/**
* Channel to receive data from
*/
interface EelReceiveChannel<out ERR : Any> {
interface EelReceiveChannel {
/**
* Reads data to [dst] but might read less (see buffer position).
* To read to the end, read until result is [ReadResult.EOF].
*
* @return Either IO [ERR] or [ReadResult] (see its doc for usage instructions)
* @return [ReadResult] (see its doc for usage instructions)
*/
@CheckReturnValue
suspend fun receive(dst: ByteBuffer): EelResult<ReadResult, ERR>
suspend fun receive(dst: ByteBuffer): ReadResult
/**
* Closes channel for receiving. You can't receive from the closed channel.
platform/eel/src/com/intellij/platform/eel/channels/EelSendChannel.kt
+7 -14
View File
@@ -1,15 +1,13 @@
// Copyright 2000-2025 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.channels
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.getOr
import org.jetbrains.annotations.CheckReturnValue
import java.nio.ByteBuffer
/**
* Consumes bytes as buffers. Each [send] writes `0` or more bytes or returns [ERR] in case if IO error
* Consumes bytes as buffers. Each [send] writes `0` or more bytes or throws an exception in case of IO error
*/
interface EelSendChannel<out ERR : Any> {
interface EelSendChannel {
/**
* Writes [src], suspends until written.
*
@@ -20,11 +18,10 @@ interface EelSendChannel<out ERR : Any> {
*
* This method is *not* thread-safe (i.e. you can't send two buffers and the same time).
*
* @return Either IO [ERR] string or success if [src] was written.
* Throws an exception in case of IO error.
*/
@EelSendApi
@CheckReturnValue
suspend fun send(src: ByteBuffer): EelResult<Unit, ERR>
suspend fun send(src: ByteBuffer)
/**
* Closes channel for sending. You can't send anything to a closed channel.
@@ -44,17 +41,13 @@ interface EelSendChannel<out ERR : Any> {
* In most cases, you need this function.
* This method is *not* thread-safe (i.e. you can't send two buffers and the same time).
*/
@CheckReturnValue
suspend fun <ERR : Any> EelSendChannel<ERR>.sendWholeBuffer(src: ByteBuffer): EelResult<Unit, ERR> {
@OptIn(EelSendApi::class)
suspend fun EelSendChannel.sendWholeBuffer(src: ByteBuffer) {
if (this is EelSendChannelCustomSendWholeBuffer) {
return sendWholeBufferCustom(src)
}
var result: EelResult<Unit, ERR>
do {
@Suppress("OPT_IN_USAGE")
result = send(src).also { it.getOr { return it } }
send(src)
}
while (src.hasRemaining())
return result
}
platform/eel/src/com/intellij/platform/eel/channels/EelSendChannelCustomSendWholeBuffer.kt
+3 -5
View File
@@ -1,14 +1,12 @@
// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.intellij.platform.eel.channels
import com.intellij.platform.eel.EelResult
import org.jetbrains.annotations.ApiStatus.Internal
import org.jetbrains.annotations.CheckReturnValue
import java.nio.ByteBuffer
// Channel might send the whole buffer by itself without external loop
@Internal
interface EelSendChannelCustomSendWholeBuffer<ERR : Any> : EelSendChannel<ERR> {
@CheckReturnValue
suspend fun sendWholeBufferCustom(src: ByteBuffer): EelResult<Unit, ERR>
}
interface EelSendChannelCustomSendWholeBuffer : EelSendChannel {
suspend fun sendWholeBufferCustom(src: ByteBuffer)
}
platform/execution-impl/src/com/intellij/execution/target/eel/EelTargetEnvironmentRequest.kt
+3 -3
View File
@@ -164,10 +164,10 @@ private class EelTargetEnvironment(override val request: EelTargetEnvironmentReq
val connection = acceptor.incomingConnections.receive()
launch {
copy(socket.consumeAsEelChannel(), connection.sendChannel).getOrThrow() // TODO: Process error
copy(socket.consumeAsEelChannel(), connection.sendChannel) // TODO: Process error
}
launch {
copy(connection.receiveChannel, socket.asEelChannel()).getOrThrow() // TODO: PRocess error
copy(connection.receiveChannel, socket.asEelChannel()) // TODO: PRocess error
}
}
@@ -285,4 +285,4 @@ private class EelTargetEnvironment(override val request: EelTargetEnvironmentReq
}
@Service
private class EelTargetScope(val scope: CoroutineScope)
private class EelTargetScope(val scope: CoroutineScope)
platform/execution-impl/src/com/intellij/execution/wsl/ijent/AbstractIjentVerificationAction.kt
+2 -4
View File
@@ -14,10 +14,8 @@ import com.intellij.openapi.diagnostic.logger
import com.intellij.openapi.diagnostic.runAndLogException
import com.intellij.openapi.project.DumbAwareAction
import com.intellij.openapi.ui.Messages
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.spawnProcess
import com.intellij.platform.eel.provider.utils.copy
import com.intellij.platform.eel.getOrThrow
import com.intellij.platform.eel.provider.utils.asEelChannel
import com.intellij.platform.ide.progress.ModalTaskOwner
import com.intellij.platform.ide.progress.TaskCancellation
@@ -80,7 +78,7 @@ abstract class AbstractIjentVerificationAction : DumbAwareAction() {
launch {
val process = ijent.exec.spawnProcess("uname", "-a").eelIt()
val stdout = ByteArrayOutputStream()
copy(process.stdout, stdout.asEelChannel()).getOrThrow() // TODO: process errors
copy(process.stdout, stdout.asEelChannel()) // TODO: process errors
withContext(Dispatchers.EDT + ModalityState.any().asContextElement()) {
Messages.showInfoMessage(stdout.toString(), title)
}
@@ -119,4 +117,4 @@ abstract class AbstractIjentVerificationAction : DumbAwareAction() {
companion object {
protected val LOG = logger<AbstractIjentVerificationAction>()
}
}
}
platform/platform-tests/testSrc/com/intellij/execution/eel/EelChannelToolsTest.kt
+81 -75
View File
@@ -3,7 +3,6 @@
package com.intellij.execution.eel
import com.intellij.platform.eel.EelResult
import com.intellij.platform.eel.ReadResult
import com.intellij.platform.eel.ReadResult.EOF
import com.intellij.platform.eel.ReadResult.NOT_EOF
@@ -11,9 +10,6 @@ import com.intellij.platform.eel.channels.EelReceiveChannel
import com.intellij.platform.eel.channels.EelSendApi
import com.intellij.platform.eel.channels.EelSendChannel
import com.intellij.platform.eel.channels.sendWholeBuffer
import com.intellij.platform.eel.getOrThrow
import com.intellij.platform.eel.provider.ResultErrImpl
import com.intellij.platform.eel.provider.ResultOkImpl
import com.intellij.platform.eel.provider.utils.*
import com.intellij.testFramework.common.timeoutRunBlocking
import io.ktor.util.decodeString
@@ -59,7 +55,7 @@ class EelChannelToolsTest {
@IntRangeSource(from = 1, to = TEXT.length + 1) srcBlockSize: Int,
): Unit = timeoutRunBlocking {
val channel = ByteArrayInputStreamLimited(data, srcBlockSize).consumeAsEelChannel()
val text = channel.readWholeText(bufferSize).getOrThrow()
val text = channel.readWholeText(bufferSize)
assertEquals(TEXT, text)
}
@@ -132,10 +128,10 @@ class EelChannelToolsTest {
val srcErrorText = "src err"
val dstErrorText = "dst err"
val src = spyk<EelReceiveChannel<IOException>>(ByteArrayInputStream(data).consumeAsEelChannel())
val src = spyk(ByteArrayInputStream(data).consumeAsEelChannel())
coEvery { src.receive(any()) } answers {
if (srcErr) {
ResultErrImpl(IOException(srcErrorText))
throw IOException(srcErrorText)
}
else callOriginal()
}
@@ -145,26 +141,24 @@ class EelChannelToolsTest {
@Suppress("OPT_IN_USAGE") dst.send(any())
} answers {
if (dstErr) {
ResultErrImpl(IOException(dstErrorText))
throw IOException(dstErrorText)
}
else callOriginal()
}
// Due to mokk bug `Default` can't be used
when (val r = copy(src, dst, dispatcher = Dispatchers.IO)) {
is EelResult.Ok -> {
assertTrue(!srcErr && !dstErr, "Error should be returned")
}
is EelResult.Error -> {
when (val r = r.error) {
is CopyResultError.InError -> {
assertTrue(srcErr, "Unexpected src error")
assertEquals(srcErrorText, r.inError.message, "Wrong src error")
}
is CopyResultError.OutError -> {
assertTrue(dstErr, "Unexpected dst error")
assertEquals(dstErrorText, r.outError.message, "Wrong dst error")
}
try {
copy(src, dst, dispatcher = Dispatchers.IO)
assertTrue(!srcErr && !dstErr, "Error should be returned")
} catch (e: CopyError) {
when (e) {
is CopyError.InError -> {
assertTrue(srcErr, "Unexpected src error")
assertEquals(srcErrorText, e.cause.message, "Wrong src error")
}
is CopyError.OutError -> {
assertTrue(dstErr, "Unexpected dst error")
assertEquals(dstErrorText, e.cause.message, "Wrong dst error")
}
}
}
@@ -173,36 +167,35 @@ class EelChannelToolsTest {
@CartesianTest
fun testErrorWithContinue(@CartesianTest.Values(booleans = [true, false]) dstError: Boolean) = timeoutRunBlocking {
val limit = 10
val src = object : EelReceiveChannel<IOException> {
val src = object : EelReceiveChannel {
var error = false
var counter = limit
override suspend fun receive(dst: ByteBuffer): EelResult<ReadResult, IOException> {
override suspend fun receive(dst: ByteBuffer): ReadResult {
error = !error
return when {
error && !dstError -> ResultErrImpl(SocketTimeoutException("wait"))
counter < 0 -> ResultOkImpl(EOF)
else -> {
dst.put(counter.toByte())
counter -= 1
ResultOkImpl(NOT_EOF)
}
if (error && !dstError) {
throw SocketTimeoutException("wait")
}
if (counter < 0) {
return EOF
}
dst.put(counter.toByte())
counter -= 1
return NOT_EOF
}
override suspend fun close() = Unit
}
val result = mutableListOf<Byte>()
val dst = object : EelSendChannel<IOException> {
val dst = object : EelSendChannel {
override val closed: Boolean = false
var error = false
@Suppress("OPT_IN_OVERRIDE")
override suspend fun send(dst: ByteBuffer): EelResult<Unit, IOException> {
override suspend fun send(dst: ByteBuffer) {
error = !error
if (error && dstError) return ResultErrImpl(SocketTimeoutException("wait"))
if (error && dstError) throw SocketTimeoutException("wait")
result.add(dst.get())
return ResultOkImpl(Unit)
}
override suspend fun close() = Unit
@@ -217,7 +210,7 @@ class EelChannelToolsTest {
assertEquals(SocketTimeoutException::class, it::class)
errorHappened = true
OnError.RETRY
}).getOrThrow()
})
assertTrue(errorHappened, "No error callback called")
assertArrayEquals((0..limit).map { it.toByte() }.reversed().toTypedArray(), result.toTypedArray(), "wrong data copied")
@@ -226,20 +219,20 @@ class EelChannelToolsTest {
@Test
fun testStreamNeverReturnsZero(): Unit = timeoutRunBlocking {
var byteCounter = 2
val channel = object : EelReceiveChannel<IOException> {
val channel = object : EelReceiveChannel {
override suspend fun receive(dst: ByteBuffer): EelResult<ReadResult, Nothing> {
override suspend fun receive(dst: ByteBuffer): ReadResult {
when {
byteCounter > 0 -> {
dst.put(42)
}
byteCounter == 0 -> Unit
byteCounter < 0 -> {
return ResultOkImpl(EOF)
return EOF
}
}
byteCounter -= 1
return ResultOkImpl(NOT_EOF)
return NOT_EOF
}
override suspend fun close() = Unit
@@ -259,7 +252,7 @@ class EelChannelToolsTest {
expect(inputStream.write(anyObject(), anyInt(), anyInt()))
expect(inputStream.flush())
replay(inputStream)
inputStream.asEelChannel().send(allocate(42)).getOrThrow()
inputStream.asEelChannel().send(allocate(42))
verify(inputStream)
}
@@ -272,35 +265,38 @@ class EelChannelToolsTest {
// 8192
launch {
pipe.sink.sendWholeBuffer(allocate(bytesCount)).getOrThrow()
pipe.sink.sendWholeBuffer(allocate(bytesCount))
} // 8192 * 2
launch {
pipe.sink.sendWholeBuffer(allocate(bytesCount)).getOrThrow()
pipe.sink.sendWholeBuffer(allocate(bytesCount))
}
awaitForCondition {
Assertions.assertEquals(bytesCount * 2, input.available(), "Wrong number of bytes available")
} // 8192
pipe.source.receive(allocate(bytesCount)).getOrThrow() // 8193
pipe.source.receive(allocate(bytesCount)) // 8193
launch {
pipe.sink.sendWholeBuffer(allocate(1)).getOrThrow()
pipe.sink.sendWholeBuffer(allocate(1))
}
awaitForCondition {
Assertions.assertEquals(bytesCount + 1, input.available(), "Wrong number of bytes available")
} // 1
pipe.source.receive(allocate(bytesCount)).getOrThrow()
pipe.source.receive(allocate(bytesCount))
awaitForCondition {
Assertions.assertEquals(1, input.available(), "After receiving there must be 0 bytes")
}
//0
pipe.source.receive(allocate(bytesCount)).getOrThrow()
pipe.source.receive(allocate(bytesCount))
awaitForCondition {
Assertions.assertEquals(0, input.available(), "After receiving there must be 0 bytes")
}
launch {
when (pipe.sink.sendWholeBuffer(allocate(bytesCount))) {
is EelResult.Error -> Unit
is EelResult.Ok -> fail("Writing to the closed channel must fail")
try {
pipe.sink.sendWholeBuffer(allocate(bytesCount))
fail("Writing to the closed channel must fail")
}
catch (e: IOException) {
// Expected exception
}
}
awaitForCondition {
@@ -326,7 +322,7 @@ class EelChannelToolsTest {
}
else {
copy(src.consumeAsEelChannel(), dst.asEelChannel(), bufferSize)
}.getOrThrow()
}
assertEquals(TEXT, dst.toByteArray().decodeToString())
}
@@ -342,7 +338,7 @@ class EelChannelToolsTest {
async {
val input = pipe.sink
repeat(repeatText) {
input.sendWholeBuffer(wrap(data)).getOrThrow()
input.sendWholeBuffer(wrap(data))
}
assertFalse(input.closed)
input.close()
@@ -350,7 +346,7 @@ class EelChannelToolsTest {
}
val buffer = allocate(blockSize)
while (pipe.source.receive(buffer).getOrThrow() != EOF) {
while (pipe.source.receive(buffer) != EOF) {
if (!buffer.hasRemaining()) {
result.put(buffer.flip())
buffer.clear()
@@ -368,7 +364,7 @@ class EelChannelToolsTest {
fun outputChannelTest(): Unit = timeoutRunBlocking {
val outputStream = ByteArrayOutputStream(data.size)
val eelChannel = outputStream.asEelChannel()
eelChannel.send(wrap(data)).getOrThrow()
eelChannel.send(wrap(data))
assertEquals(TEXT, outputStream.toByteArray().decodeToString())
}
@@ -379,7 +375,7 @@ class EelChannelToolsTest {
val inputStream = ByteArrayInputStreamLimited(data, srcBlockSize)
val eelChannel = inputStream.consumeAsEelChannel()
val buffer = allocate(data.size)
while (buffer.hasRemaining() && eelChannel.receive(buffer).getOrThrow() != EOF) {
while (buffer.hasRemaining() && eelChannel.receive(buffer) != EOF) {
}
val textFromChannel = buffer.rewind().decodeString()
assertEquals(TEXT, textFromChannel)
@@ -390,9 +386,9 @@ class EelChannelToolsTest {
@IntRangeSource(from = 1, to = TEXT.length + 1) srcBlockSize: Int,
): Unit = timeoutRunBlocking {
val eelChannel = ByteArrayInputStreamLimited(data, srcBlockSize).consumeAsEelChannel()
val result = eelChannel.readAllBytes().getOrThrow().decodeToString()
val result = eelChannel.readAllBytes().decodeToString()
assertEquals(TEXT, result)
assertEquals(0, eelChannel.readAllBytes().getOrThrow().size)
assertEquals(0, eelChannel.readAllBytes().size)
}
@CartesianTest
@@ -400,7 +396,7 @@ class EelChannelToolsTest {
@IntRangeSource(from = 1, to = TEXT.length + 1) srcBlockSize: Int,
): Unit = timeoutRunBlocking {
val eelChannel = ByteArrayInputStreamLimited(data, srcBlockSize).consumeAsEelChannel()
assertEquals(NOT_EOF, eelChannel.receive(allocate(0)).getOrThrow())
assertEquals(NOT_EOF, eelChannel.receive(allocate(0)))
assertEquals(0, eelChannel.consumeAsInputStream().read(ByteArray(data.size), data.size, 0))
}
@@ -408,9 +404,12 @@ class EelChannelToolsTest {
fun testPipeWithErrorClosed(): Unit = timeoutRunBlocking {
val pipe = EelPipe()
pipe.source.close()
when (pipe.sink.sendWholeText("D")) {
is EelResult.Ok -> fail("Writing into closed channel must be an error")
is EelResult.Error -> Unit
try {
pipe.sink.send(ByteBuffer.wrap("D".toByteArray()))
fail("Writing into closed channel must be an error")
}
catch (e: IOException) {
// Expected exception
}
}
@@ -422,13 +421,20 @@ class EelChannelToolsTest {
val expectedMessageError = "Pipe was broken with message: ${error.message}"
pipe.closePipe(error)
when (val r = pipe.sink.sendWholeText("D")) {
is EelResult.Ok -> fail("Writing into broken pipe must be an error")
is EelResult.Error -> assertEquals(expectedMessageError, r.error.message)
try {
pipe.sink.send(ByteBuffer.wrap("D".toByteArray()))
fail("Writing into broken pipe must be an error")
}
when (val r = pipe.source.receive(allocate(1))) {
is EelResult.Ok -> fail("Reading from broken pipe must be an error")
is EelResult.Error -> assertEquals(expectedMessageError, r.error.message)
catch (e: IOException) {
assertEquals(expectedMessageError, e.message)
}
try {
pipe.source.receive(allocate(1))
fail("Reading from broken pipe must be an error")
}
catch (e: IOException) {
assertEquals(expectedMessageError, e.message)
}
}
@@ -438,13 +444,13 @@ class EelChannelToolsTest {
val pipe = EelPipe()
val sendJob1 = launch {
for (c in 'a'..'z') {
pipe.sink.sendWholeText("$c").getOrThrow()
pipe.sink.sendWholeText("$c")
lettersSent.add(c)
}
}
val sendJob2 = launch {
for (c in 'A'..'Z') {
pipe.sink.sendWholeText("$c").getOrThrow()
pipe.sink.sendWholeText("$c")
lettersSent.add(c)
}
}
@@ -456,7 +462,7 @@ class EelChannelToolsTest {
sendJob2.join()
pipe.sink.close()
assertTrue(pipe.sink.closed)
val text = readJob.await().getOrThrow()
val text = readJob.await()
assertThat("Some litters missing", text.toCharArray().toList(), containsInAnyOrder(*lettersSent.toTypedArray()))
}
@@ -479,9 +485,9 @@ class EelChannelToolsTest {
@Test
fun testKotlinChannelWithError(
): Unit = timeoutRunBlocking {
val brokenChannel = mockk<EelReceiveChannel<IOException>>()
val brokenChannel = mockk<EelReceiveChannel>()
val error = IOException("go away me busy")
coEvery { brokenChannel.receive(any()) } returns ResultErrImpl(error)
coEvery { brokenChannel.receive(any()) } answers { throw error }
try {
consumeReceiveChannelAsKotlin(brokenChannel).receive()
}
@@ -506,7 +512,7 @@ class EelChannelToolsTest {
val eelChannel = ByteArrayInputStreamLimited(lines.joinToString(separator).encodeToByteArray(), blockSize).consumeAsEelChannel()
val result = mutableListOf<String>()
eelChannel.lines().collect { line ->
result.add(line.getOrThrow().trim())
result.add(line.trim())
}
Assertions.assertArrayEquals(lines.toTypedArray(), result.toTypedArray(), "Wrong lines collected")
}
@@ -566,4 +572,4 @@ private suspend fun awaitForCondition(code: suspend () -> Unit) {
}
}
code()
}
}
platform/platform-tests/testSrc/com/intellij/execution/eel/EelLocalExecApiTest.kt
+5 -5
View File
@@ -127,7 +127,7 @@ class EelLocalExecApiTest {
else {
process.stdout // stderr is redirected to stdout when launched with PTY
}
while (helloStream.receive(text).getOrThrow() != ReadResult.EOF) {
while (helloStream.receive(text) != ReadResult.EOF) {
if (HELLO in text.slice(0, text.position()).decodeString()) break
}
}
@@ -140,7 +140,7 @@ class EelLocalExecApiTest {
var ttyState: TTYState? = null
text.clear()
while (ttyState == null) {
process.stdout.receive(text).getOrThrow()
process.stdout.receive(text)
// tty might insert "\r\n", we need to remove them, hence, NEW_LINES.
// Schlemiel the Painter's Algorithm is OK in tests: do not use in production
ttyState = TTYState.deserializeIfValid(text.slice(0, text.position()).decodeString().replace(NEW_LINES, ""))
@@ -224,7 +224,7 @@ class EelLocalExecApiTest {
*
* @see `com.pty4j.PtyProcessBuilder.setUnixOpenTtyToPreserveOutputAfterTermination`
*/
private fun EelReceiveChannel<IOException>.readAllBytesAsync(coroutineScope: CoroutineScope) {
private fun EelReceiveChannel.readAllBytesAsync(coroutineScope: CoroutineScope) {
coroutineScope.launch {
readAllBytes()
}
@@ -234,6 +234,6 @@ class EelLocalExecApiTest {
* Sends [command] to the helper and flush
*/
private suspend fun EelProcess.sendCommand(command: Command) {
stdin.sendWholeText(command.name + "\n").getOrThrow()
stdin.sendWholeText(command.name + "\n")
}
}
}
platform/platform-tests/testSrc/com/intellij/execution/eel/EelLocalTunnelApiTest.kt
+7 -7
View File
@@ -60,10 +60,10 @@ class EelLocalTunnelApiTest {
fun testClientSuccessConnection(): Unit = timeoutRunBlocking(1.minutes) {
withServer { connection, _ ->
val buffer = ByteBuffer.allocate(4096)
connection.receiveChannel.receive(buffer).getOrThrow()
connection.receiveChannel.receive(buffer)
Assertions.assertEquals(NetworkConstants.HELLO_FROM_SERVER, NetworkConstants.fromByteBuffer(buffer.flip()))
connection.sendChannel.sendWholeBuffer(NetworkConstants.HELLO_FROM_CLIENT.toBuffer()).getOrThrow() // Helper closes the stream, so does the channel
Assertions.assertEquals(ReadResult.EOF, connection.receiveChannel.receive(ByteBuffer.allocate(1)).getOrThrow())
connection.sendChannel.sendWholeBuffer(NetworkConstants.HELLO_FROM_CLIENT.toBuffer()) // Helper closes the stream, so does the channel
Assertions.assertEquals(ReadResult.EOF, connection.receiveChannel.receive(ByteBuffer.allocate(1)))
}
}
@@ -71,11 +71,11 @@ class EelLocalTunnelApiTest {
fun testServerListensForConnection(): Unit = timeoutRunBlocking(1.minutes) {
val helper = clientExecutor.createBuilderToExecuteMain(localEel.exec).eelIt()
val acceptor = localEel.tunnels.getAcceptorForRemotePort().eelIt()
helper.stdin.sendWholeText(acceptor.boundAddress.port.toString() + "\n").getOrThrow()
helper.stdin.sendWholeText(acceptor.boundAddress.port.toString() + "\n")
val conn = acceptor.incomingConnections.receive()
try {
val buff = ByteBuffer.allocate(1024)
conn.receiveChannel.receive(buff).getOrThrow()
conn.receiveChannel.receive(buff)
val fromServer = NetworkConstants.fromByteBuffer(buff.flip())
assertEquals(NetworkConstants.HELLO_FROM_CLIENT, fromServer)
}
@@ -111,7 +111,7 @@ class EelLocalTunnelApiTest {
}
}
}
tx.sendWholeBuffer(ByteBuffer.wrap(helloFromServer)).getOrThrow()
tx.sendWholeBuffer(ByteBuffer.wrap(helloFromServer))
val bufferRecv = ByteBuffer.allocate(helloFromClient.size)
while (bufferRecv.hasRemaining()) {
rx.receive(bufferRecv)
@@ -137,4 +137,4 @@ class EelLocalTunnelApiTest {
helper.kill()
}
}
}
}
python/python-exec-service/src/com/intellij/python/community/execService/interactiveApi.kt
+1 -1
View File
@@ -33,7 +33,7 @@ fun interface ProcessInteractiveHandler<T> {
/**
* Process stdout -> result
*/
typealias ProcessSemiInteractiveFun<T> = suspend (EelSendChannel<IOException>, Deferred<Int>) -> Result<T, CustomErrorMessage?>
typealias ProcessSemiInteractiveFun<T> = suspend (EelSendChannel, Deferred<Int>) -> Result<T, CustomErrorMessage?>
/**
* [ProcessInteractiveHandler], but you do not have to collect output by yourself. You only have access to stdout and exit code.
python/python-exec-service/tests/com/intellij/python/junit5Tests/unit/alsoWin/ExecServiceShowCaseTest.kt
+6 -6
View File
@@ -101,10 +101,10 @@ class ExecServiceShowCaseTest {
val shell = eelHolder.eel.exec.getShell().first
val output = ExecService().executeInteractive(WhatToExec.Binary(shell.asNioPath()), emptyList(), processInteractiveHandler = ProcessInteractiveHandler<String> { _, _, process ->
val stdout = async {
process.stdout.readWholeText().getOrThrow()
process.stdout.readWholeText()
}
process.stdin.sendWholeText("echo $string\n").getOrThrow()
process.stdin.sendWholeText("exit\n").getOrThrow()
process.stdin.sendWholeText("echo $string\n")
process.stdin.sendWholeText("exit\n")
Result.success(stdout.await())
}).orThrow()
assertThat("No expected output", output, CoreMatchers.containsString(string))
@@ -118,7 +118,7 @@ class ExecServiceShowCaseTest {
val messageToUser = "abc123"
val shell = eelHolder.eel.exec.getShell().first
val result = ExecService().executeInteractive(WhatToExec.Binary(shell.asNioPath()), emptyList(), processInteractiveHandler = processSemiInteractiveHandler<Unit> { channel, exitCode ->
channel.sendWholeText("exit\n").getOrThrow()
channel.sendWholeText("exit\n")
assertEquals(0, exitCode.await(), "Wrong exit code")
if (sunny) {
Result.success(Unit)
@@ -188,10 +188,10 @@ class ExecServiceShowCaseTest {
ExecService().executeInteractive(whatToExec, args = emptyList(), processInteractiveHandler = processSemiInteractiveHandler<Unit>(progressCapturer) { stdin, exitCode ->
for (string in text) {
stdin.sendWholeText("echo $string\n").getOrThrow()
stdin.sendWholeText("echo $string\n")
delay(500)
}
stdin.sendWholeText("exit\n").getOrThrow()
stdin.sendWholeText("exit\n")
Result.success(Unit)
}).getOrThrow()
assertThat("Progress lost", progress, Matchers.containsInRelativeOrder(*text))
python/python-hatch/src/com/intellij/python/hatch/cli/HatchCli.kt
+6 -1
View File
@@ -16,6 +16,7 @@ import com.jetbrains.python.errorProcessing.PyExecResult
import com.jetbrains.python.errorProcessing.PyResult
import io.github.z4kn4fein.semver.Version
import io.github.z4kn4fein.semver.VersionFormatException
import java.io.IOException
import java.nio.file.Path
/**
@@ -125,7 +126,11 @@ class HatchCli(private val runtime: HatchRuntime) {
).makeOptions()
return runtime.executeInteractive("new", *options) { eelProcess, _ ->
if (initExistingProject) {
eelProcess.sendWholeText("$projectName\n").getOr { return@executeInteractive Result.failure("Failed to write to process: ${it.error.localizedMessage}") }
try {
eelProcess.sendWholeText("$projectName\n")
} catch (error: IOException) {
return@executeInteractive Result.failure("Failed to write to process: ${error.localizedMessage}")
}
}
Result.success("Created")
}
python/python-terminal/tests/com/intellij/python/junit5Tests/env/terminal/PyVirtualEnvTerminalCustomizerTest.kt
Vendored
+3 -3
View File
@@ -145,11 +145,11 @@ class PyVirtualEnvTerminalCustomizerTest {
val process = execOptions.eelIt()
try {
val stderr = async {
process.stderr.readWholeText().getOrThrow()
process.stderr.readWholeText()
}
val stdout = async {
val separator = if (SystemInfo.isWindows) "\n" else "\r\n"
process.stdout.readWholeText().getOrThrow().split(separator).map { it.trim() }
process.stdout.readWholeText().split(separator).map { it.trim() }
}
// tool -- where.exe Windows, "type(1)" **nix
@@ -162,7 +162,7 @@ class PyVirtualEnvTerminalCustomizerTest {
val quot = if (shellType == ShellType.ZSH) "'" else ""
Triple("type", "python is $quot", quot)
}
process.stdin.sendWholeText("$locateTool python\nexit\n").getOrThrow()
process.stdin.sendWholeText("$locateTool python\nexit\n")
val error = stderr.await()
Assertions.assertTrue(error.isEmpty(), "Unexpected text in stderr: $error")
python/services/system-python/tests/com/intellij/python/junit5Tests/env/systemPython/SystemPythonServiceShowCaseTest.kt
Vendored
+1 -1
View File
@@ -50,7 +50,7 @@ class SystemPythonServiceShowCaseTest {
val eelApi = systemPython.pythonBinary.getEelDescriptor().upgrade()
val process = eelApi.exec.spawnProcess(systemPython.pythonBinary.pathString, "--version").eelIt()
val output = async {
(if (systemPython.languageLevel.isPy3K) process.stdout else process.stderr).readWholeText().getOrThrow()
(if (systemPython.languageLevel.isPy3K) process.stdout else process.stderr).readWholeText()
}
Assertions.assertEquals(0, process.exitCode.await(), "Wrong exit code")
val versionString = PythonSdkFlavor.getLanguageLevelFromVersionStringStaticSafe(output.await())!!