mirror of
https://github.com/nushell/nushell.git
synced 2025-08-12 07:19:12 +02:00
Split nu-cli into nu-cli/nu-engine (#2898)
We split off the evaluation engine part of nu-cli into its own crate. This helps improve build times for nu-cli by 17% in my tests. It also helps us see a bit better what's the core engine portion vs the part specific to the interactive CLI piece. There's more than can be done here, but I think it's a good start in the right direction.
This commit is contained in:
Jonathan Turner
GitHub
172
crates/nu-engine/src/plugin/build_plugin.rs
Normal file
172
crates/nu-engine/src/plugin/build_plugin.rs
Normal file
@ -0,0 +1,172 @@
|
||||
use crate::plugin::run_plugin::PluginCommandBuilder;
|
||||
use log::trace;
|
||||
use nu_errors::ShellError;
|
||||
use nu_plugin::jsonrpc::JsonRpc;
|
||||
use nu_protocol::{Signature, Value};
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
|
||||
use rayon::prelude::*;
|
||||
|
||||
pub fn build_plugin_command(
|
||||
path: &std::path::Path,
|
||||
) -> Result<Option<PluginCommandBuilder>, ShellError> {
|
||||
let ext = path.extension();
|
||||
let ps1_file = match ext {
|
||||
Some(ext) => ext == "ps1",
|
||||
None => false,
|
||||
};
|
||||
|
||||
let mut child: Child = if ps1_file {
|
||||
Command::new("pwsh")
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.args(&[
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-File",
|
||||
&path.to_string_lossy(),
|
||||
])
|
||||
.spawn()
|
||||
.expect("Failed to spawn PowerShell process")
|
||||
} else {
|
||||
Command::new(path)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.spawn()
|
||||
.expect("Failed to spawn child process")
|
||||
};
|
||||
|
||||
let stdin = child.stdin.as_mut().expect("Failed to open stdin");
|
||||
let stdout = child.stdout.as_mut().expect("Failed to open stdout");
|
||||
|
||||
let mut reader = BufReader::new(stdout);
|
||||
|
||||
let request = JsonRpc::new("config", Vec::<Value>::new());
|
||||
let request_raw = serde_json::to_string(&request)?;
|
||||
trace!(target: "nu::load", "plugin infrastructure config -> path {:#?}, request {:?}", &path, &request_raw);
|
||||
stdin.write_all(format!("{}\n", request_raw).as_bytes())?;
|
||||
let path = dunce::canonicalize(path)?;
|
||||
|
||||
let mut input = String::new();
|
||||
let result = match reader.read_line(&mut input) {
|
||||
Ok(count) => {
|
||||
trace!(target: "nu::load", "plugin infrastructure -> config response for {:#?}", &path);
|
||||
trace!(target: "nu::load", "plugin infrastructure -> processing response ({} bytes)", count);
|
||||
trace!(target: "nu::load", "plugin infrastructure -> response: {}", input);
|
||||
|
||||
let response = serde_json::from_str::<JsonRpc<Result<Signature, ShellError>>>(&input);
|
||||
match response {
|
||||
Ok(jrpc) => match jrpc.params {
|
||||
Ok(params) => {
|
||||
let fname = path.to_string_lossy();
|
||||
|
||||
trace!(target: "nu::load", "plugin infrastructure -> processing {:?}", params);
|
||||
|
||||
let name = params.name.clone();
|
||||
|
||||
let fname = fname.to_string();
|
||||
|
||||
Ok(Some(PluginCommandBuilder::new(&name, &fname, params)))
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
},
|
||||
Err(e) => {
|
||||
trace!(target: "nu::load", "plugin infrastructure -> incompatible {:?}", input);
|
||||
Err(ShellError::untagged_runtime_error(format!(
|
||||
"Error: {:?}",
|
||||
e
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => Err(ShellError::untagged_runtime_error(format!(
|
||||
"Error: {:?}",
|
||||
e
|
||||
))),
|
||||
};
|
||||
|
||||
let _ = child.wait();
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
pub fn scan(
|
||||
paths: Vec<std::path::PathBuf>,
|
||||
) -> Result<Vec<crate::whole_stream_command::Command>, ShellError> {
|
||||
let mut plugins = vec![];
|
||||
|
||||
let opts = glob::MatchOptions {
|
||||
case_sensitive: false,
|
||||
require_literal_separator: false,
|
||||
require_literal_leading_dot: false,
|
||||
};
|
||||
|
||||
for path in paths {
|
||||
let mut pattern = path.to_path_buf();
|
||||
|
||||
pattern.push(std::path::Path::new("nu_plugin_[a-z0-9][a-z0-9]*"));
|
||||
|
||||
let plugs: Vec<_> = glob::glob_with(&pattern.to_string_lossy(), opts)?
|
||||
.filter_map(|x| x.ok())
|
||||
.collect();
|
||||
|
||||
let plugs: Vec<_> = plugs
|
||||
.par_iter()
|
||||
.filter_map(|path| {
|
||||
let bin_name = {
|
||||
if let Some(name) = path.file_name() {
|
||||
name.to_str().unwrap_or("")
|
||||
} else {
|
||||
""
|
||||
}
|
||||
};
|
||||
|
||||
// allow plugins with extensions on all platforms
|
||||
let is_valid_name = {
|
||||
bin_name
|
||||
.chars()
|
||||
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '.')
|
||||
};
|
||||
|
||||
let is_executable = {
|
||||
#[cfg(windows)]
|
||||
{
|
||||
bin_name.ends_with(".exe")
|
||||
|| bin_name.ends_with(".bat")
|
||||
|| bin_name.ends_with(".cmd")
|
||||
|| bin_name.ends_with(".py")
|
||||
|| bin_name.ends_with(".ps1")
|
||||
}
|
||||
|
||||
#[cfg(not(windows))]
|
||||
{
|
||||
!bin_name.contains('.')
|
||||
|| (bin_name.ends_with('.')
|
||||
|| bin_name.ends_with(".py")
|
||||
|| bin_name.ends_with(".rb")
|
||||
|| bin_name.ends_with(".sh")
|
||||
|| bin_name.ends_with(".bash")
|
||||
|| bin_name.ends_with(".zsh")
|
||||
|| bin_name.ends_with(".pl")
|
||||
|| bin_name.ends_with(".awk")
|
||||
|| bin_name.ends_with(".ps1"))
|
||||
}
|
||||
};
|
||||
|
||||
if is_valid_name && is_executable {
|
||||
trace!(target: "nu::load", "plugin infrastructure -> Trying {:?}", path.display());
|
||||
build_plugin_command(&path).unwrap_or(None)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}).map(|p| p.build())
|
||||
.filter_map(Result::ok)
|
||||
.collect::<Vec<crate::whole_stream_command::Command>>();
|
||||
plugins.extend(plugs);
|
||||
}
|
||||
|
||||
Ok(plugins)
|
||||
}
|
||||
2
crates/nu-engine/src/plugin/mod.rs
Normal file
2
crates/nu-engine/src/plugin/mod.rs
Normal file
@ -0,0 +1,2 @@
|
||||
pub mod build_plugin;
|
||||
pub(crate) mod run_plugin;
|
||||
451
crates/nu-engine/src/plugin/run_plugin.rs
Normal file
451
crates/nu-engine/src/plugin/run_plugin.rs
Normal file
@ -0,0 +1,451 @@
|
||||
use crate::command_args::CommandArgs;
|
||||
use crate::whole_stream_command::{whole_stream_command, WholeStreamCommand};
|
||||
use async_trait::async_trait;
|
||||
use derive_new::new;
|
||||
use futures::StreamExt;
|
||||
use log::trace;
|
||||
use nu_errors::ShellError;
|
||||
use nu_plugin::jsonrpc::JsonRpc;
|
||||
use nu_protocol::{Primitive, ReturnValue, Signature, UntaggedValue, Value};
|
||||
use nu_stream::{OutputStream, ToOutputStream};
|
||||
use serde::{self, Deserialize, Serialize};
|
||||
use std::collections::VecDeque;
|
||||
use std::io::prelude::*;
|
||||
use std::io::BufReader;
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
use std::process::{Child, Command, Stdio};
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
#[serde(tag = "method")]
|
||||
#[allow(non_camel_case_types)]
|
||||
pub enum NuResult {
|
||||
response {
|
||||
params: Result<VecDeque<ReturnValue>, ShellError>,
|
||||
},
|
||||
}
|
||||
|
||||
enum PluginCommand {
|
||||
Filter(PluginFilter),
|
||||
Sink(PluginSink),
|
||||
}
|
||||
|
||||
impl PluginCommand {
|
||||
fn command(self) -> Result<crate::whole_stream_command::Command, ShellError> {
|
||||
match self {
|
||||
PluginCommand::Filter(cmd) => Ok(whole_stream_command(cmd)),
|
||||
PluginCommand::Sink(cmd) => Ok(whole_stream_command(cmd)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
enum PluginMode {
|
||||
Filter,
|
||||
Sink,
|
||||
}
|
||||
|
||||
pub struct PluginCommandBuilder {
|
||||
mode: PluginMode,
|
||||
name: String,
|
||||
path: String,
|
||||
config: Signature,
|
||||
}
|
||||
|
||||
impl PluginCommandBuilder {
|
||||
pub fn new(
|
||||
name: impl Into<String>,
|
||||
path: impl Into<String>,
|
||||
config: impl Into<Signature>,
|
||||
) -> Self {
|
||||
let config = config.into();
|
||||
|
||||
PluginCommandBuilder {
|
||||
mode: if config.is_filter {
|
||||
PluginMode::Filter
|
||||
} else {
|
||||
PluginMode::Sink
|
||||
},
|
||||
name: name.into(),
|
||||
path: path.into(),
|
||||
config,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn build(&self) -> Result<crate::whole_stream_command::Command, ShellError> {
|
||||
let mode = &self.mode;
|
||||
|
||||
let name = self.name.clone();
|
||||
let path = self.path.clone();
|
||||
let config = self.config.clone();
|
||||
|
||||
let cmd = match mode {
|
||||
PluginMode::Filter => PluginCommand::Filter(PluginFilter { name, path, config }),
|
||||
PluginMode::Sink => PluginCommand::Sink(PluginSink { name, path, config }),
|
||||
};
|
||||
|
||||
cmd.command()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(new)]
|
||||
pub struct PluginFilter {
|
||||
name: String,
|
||||
path: String,
|
||||
config: Signature,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl WholeStreamCommand for PluginFilter {
|
||||
fn name(&self) -> &str {
|
||||
&self.name
|
||||
}
|
||||
|
||||
fn signature(&self) -> Signature {
|
||||
self.config.clone()
|
||||
}
|
||||
|
||||
fn usage(&self) -> &str {
|
||||
&self.config.usage
|
||||
}
|
||||
|
||||
async fn run(&self, args: CommandArgs) -> Result<OutputStream, ShellError> {
|
||||
run_filter(self.path.clone(), (args)).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_filter(path: String, args: CommandArgs) -> Result<OutputStream, ShellError> {
|
||||
trace!("filter_plugin :: {}", path);
|
||||
|
||||
let bos = futures::stream::iter(vec![
|
||||
UntaggedValue::Primitive(Primitive::BeginningOfStream).into_untagged_value()
|
||||
]);
|
||||
let eos = futures::stream::iter(vec![
|
||||
UntaggedValue::Primitive(Primitive::EndOfStream).into_untagged_value()
|
||||
]);
|
||||
|
||||
let args = args.evaluate_once().await?;
|
||||
|
||||
let real_path = Path::new(&path);
|
||||
let ext = real_path.extension();
|
||||
let ps1_file = match ext {
|
||||
Some(ext) => ext == "ps1",
|
||||
None => false,
|
||||
};
|
||||
|
||||
let mut child: Child = if ps1_file {
|
||||
Command::new("pwsh")
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.args(&[
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-File",
|
||||
&real_path.to_string_lossy(),
|
||||
])
|
||||
.spawn()
|
||||
.expect("Failed to spawn PowerShell process")
|
||||
} else {
|
||||
Command::new(path)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.spawn()
|
||||
.expect("Failed to spawn child process")
|
||||
};
|
||||
|
||||
let call_info = args.call_info.clone();
|
||||
|
||||
trace!("filtering :: {:?}", call_info);
|
||||
|
||||
Ok(bos
|
||||
.chain(args.input)
|
||||
.chain(eos)
|
||||
.map(move |item| {
|
||||
match item {
|
||||
Value {
|
||||
value: UntaggedValue::Primitive(Primitive::BeginningOfStream),
|
||||
..
|
||||
} => {
|
||||
// Beginning of the stream
|
||||
let stdin = child.stdin.as_mut().expect("Failed to open stdin");
|
||||
let stdout = child.stdout.as_mut().expect("Failed to open stdout");
|
||||
|
||||
let mut reader = BufReader::new(stdout);
|
||||
|
||||
let request = JsonRpc::new("begin_filter", call_info.clone());
|
||||
let request_raw = serde_json::to_string(&request);
|
||||
trace!("begin_filter:request {:?}", &request_raw);
|
||||
|
||||
match request_raw {
|
||||
Err(_) => {
|
||||
return OutputStream::one(Err(ShellError::labeled_error(
|
||||
"Could not load json from plugin",
|
||||
"could not load json from plugin",
|
||||
&call_info.name_tag,
|
||||
)));
|
||||
}
|
||||
Ok(request_raw) => {
|
||||
match stdin.write(format!("{}\n", request_raw).as_bytes()) {
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
return OutputStream::one(Err(ShellError::unexpected(
|
||||
format!("{}", err),
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut input = String::new();
|
||||
match reader.read_line(&mut input) {
|
||||
Ok(_) => {
|
||||
let response = serde_json::from_str::<NuResult>(&input);
|
||||
trace!("begin_filter:response {:?}", &response);
|
||||
|
||||
match response {
|
||||
Ok(NuResult::response { params }) => match params {
|
||||
Ok(params) => futures::stream::iter(params).to_output_stream(),
|
||||
Err(e) => futures::stream::iter(vec![ReturnValue::Err(e)])
|
||||
.to_output_stream(),
|
||||
},
|
||||
|
||||
Err(e) => OutputStream::one(Err(
|
||||
ShellError::untagged_runtime_error(format!(
|
||||
"Error while processing begin_filter response: {:?} {}",
|
||||
e, input
|
||||
)),
|
||||
)),
|
||||
}
|
||||
}
|
||||
Err(e) => OutputStream::one(Err(ShellError::untagged_runtime_error(
|
||||
format!("Error while reading begin_filter response: {:?}", e),
|
||||
))),
|
||||
}
|
||||
}
|
||||
Value {
|
||||
value: UntaggedValue::Primitive(Primitive::EndOfStream),
|
||||
..
|
||||
} => {
|
||||
// post stream contents
|
||||
let stdin = child.stdin.as_mut().expect("Failed to open stdin");
|
||||
let stdout = child.stdout.as_mut().expect("Failed to open stdout");
|
||||
|
||||
let mut reader = BufReader::new(stdout);
|
||||
|
||||
let request: JsonRpc<std::vec::Vec<Value>> = JsonRpc::new("end_filter", vec![]);
|
||||
let request_raw = serde_json::to_string(&request);
|
||||
trace!("end_filter:request {:?}", &request_raw);
|
||||
|
||||
match request_raw {
|
||||
Err(_) => {
|
||||
return OutputStream::one(Err(ShellError::labeled_error(
|
||||
"Could not load json from plugin",
|
||||
"could not load json from plugin",
|
||||
&call_info.name_tag,
|
||||
)));
|
||||
}
|
||||
Ok(request_raw) => {
|
||||
match stdin.write(format!("{}\n", request_raw).as_bytes()) {
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
return OutputStream::one(Err(ShellError::unexpected(
|
||||
format!("{}", err),
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut input = String::new();
|
||||
let stream = match reader.read_line(&mut input) {
|
||||
Ok(_) => {
|
||||
let response = serde_json::from_str::<NuResult>(&input);
|
||||
trace!("end_filter:response {:?}", &response);
|
||||
|
||||
match response {
|
||||
Ok(NuResult::response { params }) => match params {
|
||||
Ok(params) => futures::stream::iter(params).to_output_stream(),
|
||||
Err(e) => futures::stream::iter(vec![ReturnValue::Err(e)])
|
||||
.to_output_stream(),
|
||||
},
|
||||
Err(e) => futures::stream::iter(vec![Err(
|
||||
ShellError::untagged_runtime_error(format!(
|
||||
"Error while processing end_filter response: {:?} {}",
|
||||
e, input
|
||||
)),
|
||||
)])
|
||||
.to_output_stream(),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
futures::stream::iter(vec![Err(ShellError::untagged_runtime_error(
|
||||
format!("Error while reading end_filter response: {:?}", e),
|
||||
))])
|
||||
.to_output_stream()
|
||||
}
|
||||
};
|
||||
|
||||
let stdin = child.stdin.as_mut().expect("Failed to open stdin");
|
||||
|
||||
let request: JsonRpc<std::vec::Vec<Value>> = JsonRpc::new("quit", vec![]);
|
||||
let request_raw = serde_json::to_string(&request);
|
||||
trace!("quit:request {:?}", &request_raw);
|
||||
|
||||
match request_raw {
|
||||
Ok(request_raw) => {
|
||||
let _ = stdin.write(format!("{}\n", request_raw).as_bytes());
|
||||
// TODO: Handle error
|
||||
}
|
||||
Err(e) => {
|
||||
return OutputStream::one(Err(ShellError::untagged_runtime_error(
|
||||
format!("Error while processing quit response: {:?}", e),
|
||||
)));
|
||||
}
|
||||
}
|
||||
let _ = child.wait();
|
||||
|
||||
stream
|
||||
}
|
||||
|
||||
v => {
|
||||
// Stream contents
|
||||
let stdin = child.stdin.as_mut().expect("Failed to open stdin");
|
||||
let stdout = child.stdout.as_mut().expect("Failed to open stdout");
|
||||
|
||||
let mut reader = BufReader::new(stdout);
|
||||
|
||||
let request = JsonRpc::new("filter", v);
|
||||
let request_raw = serde_json::to_string(&request);
|
||||
trace!("filter:request {:?}", &request_raw);
|
||||
|
||||
match request_raw {
|
||||
Ok(request_raw) => {
|
||||
let _ = stdin.write(format!("{}\n", request_raw).as_bytes());
|
||||
// TODO: Handle error
|
||||
}
|
||||
Err(e) => {
|
||||
return OutputStream::one(Err(ShellError::untagged_runtime_error(
|
||||
format!("Error while processing filter response: {:?}", e),
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
let mut input = String::new();
|
||||
match reader.read_line(&mut input) {
|
||||
Ok(_) => {
|
||||
let response = serde_json::from_str::<NuResult>(&input);
|
||||
trace!("filter:response {:?}", &response);
|
||||
|
||||
match response {
|
||||
Ok(NuResult::response { params }) => match params {
|
||||
Ok(params) => futures::stream::iter(params).to_output_stream(),
|
||||
Err(e) => futures::stream::iter(vec![ReturnValue::Err(e)])
|
||||
.to_output_stream(),
|
||||
},
|
||||
Err(e) => OutputStream::one(Err(
|
||||
ShellError::untagged_runtime_error(format!(
|
||||
"Error while processing filter response: {:?}\n== input ==\n{}",
|
||||
e, input
|
||||
)),
|
||||
)),
|
||||
}
|
||||
}
|
||||
Err(e) => OutputStream::one(Err(ShellError::untagged_runtime_error(
|
||||
format!("Error while reading filter response: {:?}", e),
|
||||
))),
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
.flatten()
|
||||
.to_output_stream())
|
||||
}
|
||||
|
||||
#[derive(new)]
|
||||
pub struct PluginSink {
|
||||
name: String,
|
||||
path: String,
|
||||
config: Signature,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl WholeStreamCommand for PluginSink {
|
||||
fn name(&self) -> &str {
|
||||
&self.name
|
||||
}
|
||||
|
||||
fn signature(&self) -> Signature {
|
||||
self.config.clone()
|
||||
}
|
||||
|
||||
fn usage(&self) -> &str {
|
||||
&self.config.usage
|
||||
}
|
||||
|
||||
async fn run(&self, args: CommandArgs) -> Result<OutputStream, ShellError> {
|
||||
run_sink(self.path.clone(), args).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_sink(path: String, args: CommandArgs) -> Result<OutputStream, ShellError> {
|
||||
let args = args.evaluate_once().await?;
|
||||
let call_info = args.call_info.clone();
|
||||
|
||||
let input: Vec<Value> = args.input.collect().await;
|
||||
|
||||
let request = JsonRpc::new("sink", (call_info.clone(), input));
|
||||
let request_raw = serde_json::to_string(&request);
|
||||
if let Ok(request_raw) = request_raw {
|
||||
if let Ok(mut tmpfile) = tempfile::NamedTempFile::new() {
|
||||
let _ = writeln!(tmpfile, "{}", request_raw);
|
||||
let _ = tmpfile.flush();
|
||||
|
||||
let real_path = Path::new(&path);
|
||||
let ext = real_path.extension();
|
||||
let ps1_file = match ext {
|
||||
Some(ext) => ext == "ps1",
|
||||
None => false,
|
||||
};
|
||||
|
||||
// TODO: This sink may not work in powershell, trying to find
|
||||
// an example of what CallInfo would look like in this temp file
|
||||
let child = if ps1_file {
|
||||
Command::new("pwsh")
|
||||
.args(&[
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-File",
|
||||
&real_path.to_string_lossy(),
|
||||
&tmpfile
|
||||
.path()
|
||||
.to_str()
|
||||
.expect("Failed getting tmpfile path"),
|
||||
])
|
||||
.spawn()
|
||||
} else {
|
||||
Command::new(path).arg(&tmpfile.path()).spawn()
|
||||
};
|
||||
|
||||
if let Ok(mut child) = child {
|
||||
let _ = child.wait();
|
||||
|
||||
Ok(OutputStream::empty())
|
||||
} else {
|
||||
Err(ShellError::untagged_runtime_error(
|
||||
"Could not create process for sink command",
|
||||
))
|
||||
}
|
||||
} else {
|
||||
Err(ShellError::untagged_runtime_error(
|
||||
"Could not open file to send sink command message",
|
||||
))
|
||||
}
|
||||
} else {
|
||||
Err(ShellError::untagged_runtime_error(
|
||||
"Could not create message to sink command",
|
||||
))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user