Tag: 流

好的方法来访问混合8/16/32位字: 我在内存中有大量的二进制数据，我需要从随机访问的字节对齐地址读取/写入数据。然而，有时我需要读/写8位字，有时候是（big-endian）16位字，有时候是（big-endian）32位字。将数据表示为ByteArray并手工实现16/32位读/写是一种天真的解决方案： class Blob (val image: ByteArray, var ptr: Int = 0) { fun readWord8(): Byte = image[ptr++] fun readWord16(): Short { val hi = readWord8().toInt() and 0xff val lo = readWord8().toInt() and 0xff return ((hi shl 8) or lo).toShort() } fun readWord32(): Int { val hi = readWord16().toLong() and 0xffff val lo = […]

阿卡流，上游不产生价值: 我有一个Akka流，我想用来启动一个会话对第三方API，然后我希望它通过在设定的时间间隔发出keepAlive请求保持该会话活着。现在我设法从流中收集第一个会话签名元素，将其收集到SinkQueue中，但似乎无法获得上游来继续生成元素，这是我的代码： val signatures = Source.single({ _: String -> sessionManager.logon(logonReq, contractVersion).signature }).keepAlive(interval, {{ signature: String -> sessionManager.keepAlive(KeepAliveRequest(), contractVersion, signature) signature }}) fun start() = signatures. conflateWithSeed({ it("") } ,{ x, f -> f(x) }). runWith(Sink.queue<String>().withAttributes(Attributes.inputBuffer(0, 1)), mat)

从Rest服务调用中检索流数据: 我正在创建一个将使用以下kotlin代码调用的webservice fun submitRequest(request: PKCS10CertificationRequest): String { // Post request to certificate signing server via http. val conn = URL("$server/api/certificate").openConnection() as HttpURLConnection conn.doOutput = true conn.requestMethod = "POST" conn.setRequestProperty("Content-Type", "application/octet-stream") conn.setRequestProperty("Client-Version", clientVersion) conn.outputStream.write(request.encoded) return when (conn.responseCode) { HTTP_OK -> IOUtils.toString(conn.inputStream, conn.charset) HTTP_FORBIDDEN -> throw IOException("Client version $clientVersion is forbidden from accessing permissioning server, please upgrade to […]

RxJava-将Observable转变为Iterator，Stream或Sequence: 我知道这打破了很多Rx规则，但我真的很喜欢RxJava-JDBC ，所以我的队友也是这样。关系数据库对于我们的工作非常重要，Rx也是如此。但是有些情况下，我们不希望作为Observable<ResultSet>发出，而只是有一个基于拉的Java 8 Stream<ResultSet>或Kotlin Sequence<ResultSet> 。但是我们非常习惯于只返回一个Observable<ResultSet>的RxJava-JDBC库。因此，我想知道是否有一种方法可以使用扩展函数将Observable<ResultSet>转换为Sequence<ResultSet> ，而不执行任何中间集合或toBlocking()调用。下面是我迄今所有的，但我的头正在旋转，试图连接推拉系统，我不能缓冲，因为每个onNext()调用ResultSet是有状态的。这是不可能的任务吗？ import rx.Observable import rx.Subscriber import java.sql.ResultSet fun Observable<ResultSet>.asSequence() = object: Iterator<ResultSet>, Subscriber<ResultSet>() { private var isComplete = false override fun onCompleted() { isComplete = true } override fun onError(e: Throwable?) { throw UnsupportedOperationException() } override fun onNext(rs: ResultSet?) { throw UnsupportedOperationException() } […]