==3752734== Memcheck, a memory error detector ==3752734== Copyright (C) 2002-2026, and GNU GPL'd, by Julian Seward et al. ==3752734== Using Valgrind-3.27.1 and LibVEX; rerun with -h for copyright info ==3752734== Command: /data/localhost/ripley/R/R-devel-vg/bin/exec/R --vanilla ==3752734== R Under development (unstable) (2026-09-17 r90559) -- "Unsuffered Consequences" Copyright (C) 2026 The R Foundation for Statistical Computing Platform: x86_64-pc-linux-gnu R is free software and comes with ABSOLUTELY NO WARRANTY. You are welcome to redistribute it under certain conditions. Type 'license()' or 'licence()' for distribution details. Natural language support but running in an English locale R is a collaborative project with many contributors. Type 'contributors()' for more information and 'citation()' on how to cite R or R packages in publications. Type 'demo()' for some demos, 'help()' for on-line help, or 'help.start()' for an HTML browser interface to help. Type 'q()' to quit R. > pkgname <- "rgrpc" > source(file.path(R.home("share"), "R", "examples-header.R")) > options(warn = 1) > library('rgrpc') > > base::assign(".oldSearch", base::search(), pos = 'CheckExEnv') > base::assign(".old_wd", base::getwd(), pos = 'CheckExEnv') > cleanEx() > nameEx("grpc_await") > ### * grpc_await > > flush(stderr()); flush(stdout()) > > ### Name: grpc_await > ### Title: Wait for events belonging to one call > ### Aliases: grpc_await > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") --3752734-- Warning: zero subprog, missing DW_AT_abstract_origin in DW_TAG_inlined_subroutine in /data/localhost/ripley/.cache/debuginfod_client/7018b287d4e48ea24c5496efe41e5259fd4c87b6/debuginfo ==3752734== Conditional jump or move depends on uninitialised value(s) ==3752734== at 0x11EA06CC: int UnwindImpl(void**, unsigned long*, int*, int, int, void const*, int*) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/internal/stacktrace_x86-inl.inc:350) ==3752734== by 0x11EA1484: Unwind (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/stacktrace.cc:125) ==3752734== by 0x11EA1484: absl::lts_20260107::GetStackTrace(void**, int, int) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/stacktrace.cc:167) ==3752734== by 0x11E9659F: absl::lts_20260107::synchronization_internal::GraphCycles::UpdateStackTrace(absl::lts_20260107::synchronization_internal::GraphId, int, int (*)(void**, int)) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/internal/graphcycles.cc:695) ==3752734== by 0x11E81438: absl::lts_20260107::DeadlockCheck(absl::lts_20260107::Mutex*) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/mutex.cc:1404) ==3752734== by 0x11E850AC: DebugOnlyDeadlockCheck (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/mutex.cc:1475) ==3752734== by 0x11E850AC: absl::lts_20260107::Mutex::lock() (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/mutex.cc:1532) ==3752734== by 0x11CA12DB: grpc_core::(anonymous namespace)::ThreadInternalsPosix::Start() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/gprpp/thd_posix.cc:167) ==3752734== by 0x117FE5C8: Start (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/gprpp/thd.h:129) ==3752734== by 0x117FE5C8: grpc_core::Executor::SetThreading(bool) (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/iomgr/executor.cc:173) ==3752734== by 0x117FF3A5: grpc_core::Executor::InitAll() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/iomgr/executor.cc:392) ==3752734== by 0x11800F1E: grpc_iomgr_init() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/iomgr/iomgr.cc:69) ==3752734== by 0x118E3EF5: grpc_init (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/surface/init.cc:160) ==3752734== by 0x11477E35: init (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/include/grpcpp/impl/grpc_library.h:34) ==3752734== by 0x11477E35: GrpcLibraryCodegen (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/include/grpcpp/impl/codegen/grpc_library.h:48) ==3752734== by 0x11477E35: GrpcLibraryCodegen (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/include/grpcpp/impl/codegen/grpc_library.h:42) ==3752734== by 0x11477E35: grpc::ServerCredentials::ServerCredentials() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/cpp/server/server_credentials.cc:25) ==3752734== by 0x1145CB69: InsecureServerCredentialsImpl (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/cpp/server/insecure_server_credentials.cc:32) ==3752734== by 0x1145CB69: grpc::InsecureServerCredentials() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/cpp/server/insecure_server_credentials.cc:54) ==3752734== Uninitialised value was created by a stack allocation ==3752734== at 0x113D5635: grpc_r_server2_create (/data/localhost/ripley/R/packages/tests-vg/rgrpc/src/server.cpp:308) ==3752734== ==3752734== Conditional jump or move depends on uninitialised value(s) ==3752734== at 0x11EA05C6: void** NextStackFrame(void**, void const*, unsigned long, unsigned long) [clone .constprop.0] (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/internal/stacktrace_x86-inl.inc:276) ==3752734== by 0x11EA06D6: int UnwindImpl(void**, unsigned long*, int*, int, int, void const*, int*) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/internal/stacktrace_x86-inl.inc:355) ==3752734== by 0x11EA1484: Unwind (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/stacktrace.cc:125) ==3752734== by 0x11EA1484: absl::lts_20260107::GetStackTrace(void**, int, int) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/debugging/stacktrace.cc:167) ==3752734== by 0x11E9659F: absl::lts_20260107::synchronization_internal::GraphCycles::UpdateStackTrace(absl::lts_20260107::synchronization_internal::GraphId, int, int (*)(void**, int)) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/internal/graphcycles.cc:695) ==3752734== by 0x11E81438: absl::lts_20260107::DeadlockCheck(absl::lts_20260107::Mutex*) (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/mutex.cc:1404) ==3752734== by 0x11E850AC: DebugOnlyDeadlockCheck (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/mutex.cc:1475) ==3752734== by 0x11E850AC: absl::lts_20260107::Mutex::lock() (/usr/src/debug/abseil-cpp-20260107.1-1.fc44.x86_64/absl/synchronization/mutex.cc:1532) ==3752734== by 0x11CA12DB: grpc_core::(anonymous namespace)::ThreadInternalsPosix::Start() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/gprpp/thd_posix.cc:167) ==3752734== by 0x117FE5C8: Start (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/gprpp/thd.h:129) ==3752734== by 0x117FE5C8: grpc_core::Executor::SetThreading(bool) (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/iomgr/executor.cc:173) ==3752734== by 0x117FF3A5: grpc_core::Executor::InitAll() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/iomgr/executor.cc:392) ==3752734== by 0x11800F1E: grpc_iomgr_init() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/iomgr/iomgr.cc:69) ==3752734== by 0x118E3EF5: grpc_init (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/core/lib/surface/init.cc:160) ==3752734== by 0x11477E35: init (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/include/grpcpp/impl/grpc_library.h:34) ==3752734== by 0x11477E35: GrpcLibraryCodegen (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/include/grpcpp/impl/codegen/grpc_library.h:48) ==3752734== by 0x11477E35: GrpcLibraryCodegen (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/include/grpcpp/impl/codegen/grpc_library.h:42) ==3752734== by 0x11477E35: grpc::ServerCredentials::ServerCredentials() (/usr/src/debug/grpc-1.48.4-57.fc44.x86_64/src/cpp/server/server_credentials.cc:25) ==3752734== Uninitialised value was created by a stack allocation ==3752734== at 0x113D5635: grpc_r_server2_create (/data/localhost/ripley/R/packages/tests-vg/rgrpc/src/server.cpp:308) ==3752734== > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > ## unary: keep waiting until the completion arrives; the call's own > ## deadline_ms is what guarantees this loop ends > call <- grpc_call(cl, "/demo.Echo/Say", charToRaw("hi"), deadline_ms = 5000) > req <- next_request(srv) > grpc_reply(req, req$request) > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break # empty just means "not yet" + } > evs[[1]]$status_name [1] "OK" > > ## server handler: drain one client-streaming call without seeing any > ## other call's messages > s <- grpc_stream(cl, "/demo.Echo/Collect", deadline_ms = 5000) > for (i in 1:3) grpc_send(s, as.raw(i)) > grpc_writes_done(s) > req <- next_request(srv) > got <- list(req$request) > repeat { + grpc_read(req) + evs <- grpc_await(req, timeout_ms = 1000L) + for (ev in evs) if (ev$type == "stream_msg") got <- c(got, list(ev$request)) + if (length(Filter(function(e) e$type %in% c("client_done", "cancelled"), evs))) break + } > grpc_reply(req, as.raw(length(got))) > length(got) [1] 3 > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_call") > ### * grpc_call > > flush(stderr()); flush(stdout()) > > ### Name: grpc_call > ### Title: Start a unary call > ### Aliases: grpc_call > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > call <- grpc_call(cl, "/demo.Echo/Say", charToRaw("ping"), deadline_ms = 5000) > class(call) [1] "grpc_call" > > ## echo it back from the server half of the same process > req <- next_request(srv) > grpc_reply(req, req$request) > > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > evs[[1]]$status_name [1] "OK" > rawToChar(evs[[1]]$response) [1] "ping" > > ## a call nobody answers ends at its deadline > call <- grpc_call(cl, "/demo.Echo/Say", raw(0), deadline_ms = 200) > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > evs[[1]]$status_name [1] "DEADLINE_EXCEEDED" > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_cancel") > ### * grpc_cancel > > flush(stderr()); flush(stdout()) > > ### Name: grpc_cancel > ### Title: Cancel an in-flight call or stream > ### Aliases: grpc_cancel > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > ## a call the server holds without answering: cancel it instead of > ## waiting for its deadline > call <- grpc_call(cl, "/demo.Echo/Say", raw(0), deadline_ms = 60000) > req <- next_request(srv) > grpc_cancel(call) > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > evs[[1]]$status_name [1] "CANCELLED" > > ## the server hears of it as a "cancelled" event, and the request can no > ## longer be answered > repeat { + evs <- grpc_await(req, timeout_ms = 1000L) + if (length(Filter(function(e) e$type == "cancelled", evs))) break + } > (grpc_reply(req, raw(0))) # FALSE [1] FALSE > (grpc_cancel(req)) # FALSE: already terminal [1] FALSE > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_client") > ### * grpc_client > > flush(stderr()); flush(stdout()) > > ### Name: grpc_client > ### Title: Create an asynchronous gRPC client > ### Aliases: grpc_client > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > grpc_state(cl) # "IDLE": nothing has connected yet [1] "IDLE" > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > call <- grpc_call(cl, "/demo.Echo/Say", as.raw(1:4), deadline_ms = 5000) > > ## answer it from the server half of the same process > req <- next_request(srv) > grpc_reply(req, req$request) > > ## completions arrive through grpc_poll(), or per call through grpc_await() > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > evs[[1]]$status_name [1] "OK" > evs[[1]]$response [1] 01 02 03 04 > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_close") > ### * grpc_close > > flush(stderr()); flush(stdout()) > > ### Name: grpc_close > ### Title: Shut down a client or server > ### Aliases: grpc_close > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > grpc_close(cl) > grpc_close(srv) > grpc_close(srv) # closing twice is a no-op > > > > cleanEx() > nameEx("grpc_decode") > ### * grpc_decode > > flush(stderr()); flush(stdout()) > > ### Name: grpc_decode > ### Title: Decode protocol buffer bytes to a message > ### Aliases: grpc_decode > > ### ** Examples > > if (requireNamespace("RProtoBuf", quietly = TRUE)) { + RProtoBuf::readProtoFiles2("health.proto", + protoPath = system.file("proto", "health", package = "rgrpc")) + ## bytes as they arrive in a request or response event + msg <- RProtoBuf::P("grpc.health.v1.HealthCheckRequest")$new(service = "demo") + bytes <- RProtoBuf::serialize(msg, NULL) + decoded <- grpc_decode(bytes, "grpc.health.v1.HealthCheckRequest") + print(decoded$service) + } [1] "demo" > > > > cleanEx() detaching ‘RProtoBuf:DescriptorPool’ > nameEx("grpc_fd") > ### * grpc_fd > > flush(stderr()); flush(stdout()) > > ### Name: grpc_fd > ### Title: Completion wakeup file descriptor > ### Aliases: grpc_fd > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > grpc_fd(cl) [1] 10 > grpc_fd(srv) [1] 5 > ## hand the descriptor to an event loop instead of polling, e.g. > ## later::later_fd(function(ready) grpc_poll(cl), readfds = grpc_fd(cl)) > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_finish") > ### * grpc_finish > > flush(stderr()); flush(stdout()) > > ### Name: grpc_finish > ### Title: End a server stream > ### Aliases: grpc_finish > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > ## server streaming: the client sends one request and half-closes > s <- grpc_stream(cl, "/demo.Echo/Watch", deadline_ms = 5000) > grpc_send(s, as.raw(7)) > grpc_writes_done(s) > > ## server: three messages, then the status with trailing metadata > req <- next_request(srv) > for (i in 1:3) grpc_send(req, as.raw(i)) > grpc_finish(req, metadata = c("x-count" = "3")) > > out <- list() > repeat { + evs <- grpc_await(s, timeout_ms = 1000L) + for (ev in evs) if (ev$kind == "stream_msg") out <- c(out, list(ev$response)) + st <- Filter(function(e) e$kind == "stream_status", evs) + if (length(st)) break + } > length(out) [1] 3 > st[[1]]$status_name [1] "OK" > st[[1]]$trailing_metadata x-count "3" > > ## an error status ends a stream without any payload > s <- grpc_stream(cl, "/demo.Echo/Watch", deadline_ms = 5000) > grpc_send(s, raw(1)) > req <- next_request(srv) > grpc_finish(req, status = "NOT_FOUND", message = "no such stream") > repeat { + evs <- grpc_await(s, timeout_ms = 1000L) + st <- Filter(function(e) e$kind == "stream_status", evs) + if (length(st)) break + } > st[[1]]$status_name [1] "NOT_FOUND" > st[[1]]$message [1] "no such stream" > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_method") > ### * grpc_method > > flush(stderr()); flush(stdout()) > > ### Name: grpc_method > ### Title: Look up a method in a resolved service > ### Aliases: grpc_method > > ### ** Examples > > if (requireNamespace("RProtoBuf", quietly = TRUE)) { + RProtoBuf::readProtoFiles2("health.proto", + protoPath = system.file("proto", "health", package = "rgrpc")) + svc <- grpc_service("grpc.health.v1.HealthCheckRequest", "Health") + m <- grpc_method(svc, "Check") + print(m$path) + print(m$input_type) + print(grpc_method(svc, "Watch")$server_streaming) + } [1] "/grpc.health.v1.Health/Check" [1] "grpc.health.v1.HealthCheckRequest" [1] TRUE > > > > cleanEx() > nameEx("grpc_pending") > ### * grpc_pending > > flush(stderr()); flush(stdout()) > > ### Name: grpc_pending > ### Title: Number of pending operations > ### Aliases: grpc_pending > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > grpc_pending(cl) # 0 [1] 0 > call <- grpc_call(cl, "/demo.Echo/Say", raw(0), deadline_ms = 5000) > grpc_pending(cl) # 1: started, not yet completed [1] 1 > > req <- next_request(srv) > grpc_pending(srv) # 1: accepted, not yet answered [1] 1 > grpc_reply(req, raw(0)) > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > grpc_pending(cl) # 0 again [1] 0 > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_poll") > ### * grpc_poll > > flush(stderr()); flush(stdout()) > > ### Name: grpc_poll > ### Title: Receive completed events > ### Aliases: grpc_poll > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## three calls in flight at once > calls <- lapply(1:3, function(i) { + grpc_call(cl, "/demo.Echo/Say", as.raw(i), deadline_ms = 5000) + }) > > ## server: one queue for every call; answer requests as they arrive > ## (5 s of silence is an error rather than an endless wait) > answered <- 0L > while (answered < 3L) { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) { + if (ev$type == "request") { + grpc_reply(ev, ev$request) + answered <- answered + 1L + } + } + } > > ## client: completions come back in completion order, so match on id > ids <- vapply(calls, function(x) x$id, numeric(1)) > got <- vector("list", length(calls)) > while (any(vapply(got, is.null, logical(1)))) { + for (ev in grpc_poll(cl, timeout_ms = 100L)) { + if (ev$kind == "unary") got[[match(ev$id, ids)]] <- ev$response + } + } > unlist(got) [1] 01 02 03 > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_read") > ### * grpc_read > > flush(stderr()); flush(stdout()) > > ### Name: grpc_read > ### Title: Pull the next inbound message on a server stream > ### Aliases: grpc_read > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > s <- grpc_stream(cl, "/demo.Echo/Collect", deadline_ms = 5000) > for (i in 1:4) grpc_send(s, as.raw(i)) > grpc_writes_done(s) > > req <- next_request(srv) > got <- as.integer(req$request) # the first message rides on the request > (grpc_read(req)) # TRUE: one read is now in flight [1] TRUE > (grpc_read(req)) # FALSE: one read at a time [1] FALSE > ## each read yields one "stream_msg", or "client_done" at the end > repeat { + evs <- grpc_await(req, timeout_ms = 1000L) + for (ev in evs) if (ev$type == "stream_msg") got <- c(got, as.integer(ev$request)) + if (length(Filter(function(e) e$type %in% c("client_done", "cancelled"), evs))) break + grpc_read(req) # post the next read + } > got [1] 1 2 3 4 > grpc_reply(req, as.raw(length(got))) > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_reply") > ### * grpc_reply > > flush(stderr()); flush(stdout()) > > ### Name: grpc_reply > ### Title: Answer an incoming request > ### Aliases: grpc_reply > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > ## a payload reply (status OK), with trailing metadata > call <- grpc_call(cl, "/demo.Echo/Say", charToRaw("hi"), deadline_ms = 5000) > req <- next_request(srv) > (grpc_reply(req, req$request, metadata = c("x-served-by" = "example"))) [1] TRUE > (grpc_reply(req, req$request)) # FALSE: already answered [1] FALSE > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > evs[[1]]$status_name [1] "OK" > evs[[1]]$trailing_metadata x-served-by "example" > > ## an error status needs no payload > call <- grpc_call(cl, "/demo.Echo/Say", raw(0), deadline_ms = 5000) > req <- next_request(srv) > grpc_reply(req, status = "NOT_FOUND", message = "no such thing") > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > evs[[1]]$status_name [1] "NOT_FOUND" > evs[[1]]$message [1] "no such thing" > is.null(evs[[1]]$response) [1] TRUE > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_send") > ### * grpc_send > > flush(stderr()); flush(stdout()) > > ### Name: grpc_send > ### Title: Send a message on a stream > ### Aliases: grpc_send > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > ## client streaming: queue five messages, then half-close > s <- grpc_stream(cl, "/demo.Echo/Collect", deadline_ms = 5000) > for (i in 1:5) grpc_send(s, as.raw(i)) > grpc_writes_done(s) > (grpc_send(s, as.raw(6))) # FALSE: no writes after the half-close [1] FALSE > > ## server: count what arrives (the first message rides on the request > ## event), then answer once, unary-style > req <- next_request(srv) > n <- 1L > repeat { + grpc_read(req) + evs <- grpc_await(req, timeout_ms = 1000L) + n <- n + length(Filter(function(e) e$type == "stream_msg", evs)) + if (length(Filter(function(e) e$type %in% c("client_done", "cancelled"), evs))) break + } > grpc_reply(req, as.raw(n)) > > ## client: the reply is one "stream_msg", followed by the "stream_status" > reply <- NULL > repeat { + evs <- grpc_await(s, timeout_ms = 1000L) + for (ev in evs) if (ev$kind == "stream_msg") reply <- ev$response + if (length(Filter(function(e) e$kind == "stream_status", evs))) break + } > as.integer(reply) [1] 5 > > ## a full write queue makes grpc_send() return FALSE; wait for the > ## "stream_writable" event (grpc_await() on the stream) and retry > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_server") > ### * grpc_server > > flush(stderr()); flush(stdout()) > > ### Name: grpc_server > ### Title: Create an asynchronous generic gRPC server > ### Aliases: grpc_server > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > grpc_server_port(srv) [1] 39019 > > ## a client in the same process makes one call > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > call <- grpc_call(cl, "/demo.Echo/Say", charToRaw("hello"), deadline_ms = 5000) > > ## the server sees it as a request event; there are no handlers to > ## register, the method name arrives with the event. One queue serves > ## every call, so step over other events; 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > req <- next_request(srv) > req$method [1] "/demo.Echo/Say" > grpc_reply(req, req$request) # echo it back > > ## the client collects the completion > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break # empty means "not yet" + } > evs[[1]]$status_name [1] "OK" > rawToChar(evs[[1]]$response) [1] "hello" > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_server_port") > ### * grpc_server_port > > flush(stderr()); flush(stdout()) > > ### Name: grpc_server_port > ### Title: Bound TCP port of a server > ### Aliases: grpc_server_port > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > grpc_server_port(srv) [1] 40785 > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_service") > ### * grpc_service > > flush(stderr()); flush(stdout()) > > ### Name: grpc_service > ### Title: Resolve a gRPC service from the RProtoBuf descriptor pool > ### Aliases: grpc_service > > ### ** Examples > > if (requireNamespace("RProtoBuf", quietly = TRUE)) { + ## the gRPC health-checking schema ships with the package + RProtoBuf::readProtoFiles2("health.proto", + protoPath = system.file("proto", "health", package = "rgrpc")) + svc <- grpc_service("grpc.health.v1.HealthCheckRequest", "Health") + print(svc$name) + print(names(svc$methods)) + } [1] "grpc.health.v1.Health" [1] "Check" "Watch" > > > > cleanEx() > nameEx("grpc_state") > ### * grpc_state > > flush(stderr()); flush(stdout()) > > ### Name: grpc_state > ### Title: Channel connectivity state > ### Aliases: grpc_state > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > grpc_state(cl) # "IDLE": channels connect on first use [1] "IDLE" > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > call <- grpc_call(cl, "/demo.Echo/Say", raw(0), deadline_ms = 5000) > req <- next_request(srv) > grpc_reply(req, raw(0)) > repeat { + evs <- grpc_await(call, timeout_ms = 1000L) + if (length(evs)) break + } > grpc_state(cl) # "READY" once a call has gone through [1] "READY" > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_stream") > ### * grpc_stream > > flush(stderr()); flush(stdout()) > > ### Name: grpc_stream > ### Title: Open a streaming call > ### Aliases: grpc_stream > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > ## client: three messages out, then half-close the request direction > s <- grpc_stream(cl, "/demo.Echo/Chat", deadline_ms = 5000) > for (i in 1:3) grpc_send(s, as.raw(i)) > grpc_writes_done(s) > > ## server: the first message rides on the request event; read the rest, > ## then echo each one back doubled and end the stream > req <- next_request(srv) > got <- list(req$request) > repeat { + grpc_read(req) + evs <- grpc_await(req, timeout_ms = 1000L) + for (ev in evs) if (ev$type == "stream_msg") got <- c(got, list(ev$request)) + if (length(Filter(function(e) e$type %in% c("client_done", "cancelled"), evs))) break + } > for (m in got) grpc_send(req, as.raw(as.integer(m) * 2L)) > grpc_finish(req) > > ## client: responses arrive as "stream_msg" events up to the "stream_status" > out <- list() > repeat { + evs <- grpc_await(s, timeout_ms = 1000L) + for (ev in evs) if (ev$kind == "stream_msg") out <- c(out, list(ev$response)) + st <- Filter(function(e) e$kind == "stream_status", evs) + if (length(st)) break + } > unlist(out) [1] 02 04 06 > st[[1]]$status_name [1] "OK" > > grpc_close(cl) > grpc_close(srv) > > > > cleanEx() > nameEx("grpc_tls") > ### * grpc_tls > > flush(stderr()); flush(stdout()) > > ### Name: grpc_tls > ### Title: TLS credentials > ### Aliases: grpc_tls > > ### ** Examples > > ## needs certificate files, which only exist in a real deployment (the > ## package's TLS tests generate throwaway ones with the openssl > ## command-line tool), so this example is not run > ## Not run: > ##D ## client pinning a CA and presenting its own identity (mTLS) > ##D creds <- grpc_tls(ca_file = "ca.pem", > ##D cert_file = "client.pem", key_file = "client.key") > ##D cl <- grpc_client("node1:41900", credentials = creds) > ##D > ##D ## server demanding client certificates signed by that CA > ##D srv <- grpc_server("0.0.0.0:41900", > ##D credentials = grpc_tls(ca_file = "ca.pem", > ##D cert_file = "server.pem", > ##D key_file = "server.key", > ##D require_client_cert = TRUE)) > ## End(Not run) > > > > cleanEx() > nameEx("grpc_version") > ### * grpc_version > > flush(stderr()); flush(stdout()) > > ### Name: grpc_version > ### Title: Version of the linked gRPC C++ library > ### Aliases: grpc_version > > ### ** Examples > > grpc_version() [1] "1.48.4" > > > > cleanEx() > nameEx("grpc_writes_done") > ### * grpc_writes_done > > flush(stderr()); flush(stdout()) > > ### Name: grpc_writes_done > ### Title: Half-close a client stream > ### Aliases: grpc_writes_done > > ### ** Examples > > srv <- grpc_server("127.0.0.1:0") > cl <- grpc_client(sprintf("127.0.0.1:%d", grpc_server_port(srv))) > > ## the next request event; the server has one queue for every call, so > ## other events are stepped over, and 5 s of silence is an error > next_request <- function(srv) { + repeat { + evs <- grpc_poll(srv, timeout_ms = 5000L) + if (!length(evs)) stop("no request within 5 s") + for (ev in evs) if (ev$type == "request") return(ev) + } + } > > s <- grpc_stream(cl, "/demo.Echo/Collect", deadline_ms = 5000) > grpc_send(s, as.raw(1)) > (grpc_writes_done(s)) # TRUE: request direction closed [1] TRUE > (grpc_writes_done(s)) # FALSE: already half-closed [1] FALSE > (grpc_send(s, as.raw(2))) # FALSE: no writes after the half-close [1] FALSE > > ## the server reads past the first message and sees "client_done" > req <- next_request(srv) > grpc_read(req) > repeat { + evs <- grpc_await(req, timeout_ms = 1000L) + done <- Filter(function(e) e$type %in% c("client_done", "cancelled"), evs) + if (length(done)) break + } > done[[1]]$type [1] "client_done" > grpc_reply(req, raw(0)) > > grpc_close(cl) > grpc_close(srv) > > > > ### *