use super::*;
use ::gclient::gear::runtime_types::pallet_gear_voucher::internal::VoucherId;
use ::gclient::{EventListener, EventProcessor as _, GearApi};
use core::task::ready;
use futures::{Stream, StreamExt, stream};
pub use gear_core_errors::{ErrorReplyReason, SimpleExecutionError};
#[derive(Debug, thiserror::Error)]
pub enum GclientError {
#[error(transparent)]
Env(#[from] gclient::Error),
#[error("reply error: {0}")]
ReplyHasError(ErrorReplyReason, crate::Vec<u8>),
#[error("reply is missing")]
ReplyIsMissing,
}
impl ReplyError for GclientError {
fn from_codec_error(err: parity_scale_codec::Error) -> Self {
gclient::Error::Codec(err).into()
}
fn userspace_panic_payload(&self) -> Option<&[u8]> {
match self {
GclientError::ReplyHasError(
ErrorReplyReason::Execution(SimpleExecutionError::UserspacePanic),
payload,
) => Some(payload),
_ => None,
}
}
}
#[derive(Clone)]
pub struct GclientEnv {
api: GearApi,
}
crate::params_struct_impl!(
GclientEnv,
GclientParams {
#[cfg(not(feature = "ethexe"))]
gas_limit: GasUnit,
value: ValueUnit,
at_block: H256,
voucher: (VoucherId, bool),
}
);
impl GclientEnv {
pub fn new(api: GearApi) -> Self {
Self { api }
}
pub fn with_suri(self, suri: impl AsRef<str>) -> Self {
let api = self.api.with(suri).unwrap();
Self { api }
}
pub async fn create_program(
&self,
code_id: CodeId,
salt: impl AsRef<[u8]>,
payload: impl AsRef<[u8]>,
params: GclientParams,
) -> Result<(ActorId, Vec<u8>), GclientError> {
let api = self.api.clone();
create_program(api, code_id, salt, payload, params).await
}
pub async fn send_for_reply(
&self,
program_id: ActorId,
payload: Vec<u8>,
params: GclientParams,
) -> Result<Vec<u8>, GclientError> {
let api = self.api.clone();
send_for_reply(api, program_id, payload, params)
.await
.map(|(_program_id, payload)| payload)
}
pub async fn send_one_way(
&self,
program_id: ActorId,
payload: Vec<u8>,
params: GclientParams,
) -> Result<MessageId, GclientError> {
send_one_way(&self.api, program_id, payload, params).await
}
pub async fn query(
&self,
destination: ActorId,
payload: impl AsRef<[u8]>,
params: GclientParams,
) -> Result<Vec<u8>, GclientError> {
query_calculate_reply(&self.api, destination, payload, params).await
}
}
impl GearEnv for GclientEnv {
type Params = GclientParams;
type Error = GclientError;
type MessageState = Pin<Box<dyn Future<Output = Result<(ActorId, Vec<u8>), GclientError>>>>;
}
impl<T: ServiceCall> PendingCall<T, GclientEnv> {
pub async fn send_one_way(&mut self) -> Result<MessageId, GclientError> {
let (payload, params) = self.take_encoded_args_and_params();
self.env
.send_one_way(self.destination, payload, params)
.await
}
pub async fn send_for_reply(mut self) -> Result<Self, GclientError> {
let (payload, params) = self.take_encoded_args_and_params();
let (listener, message_id) =
send_for_reply_and_listen(&self.env.api, self.destination, payload, params).await?;
self.state = Some(Box::pin(wait_for_reply_owned(
listener,
message_id,
self.destination,
)));
Ok(self)
}
pub async fn query(mut self) -> Result<T::Output, GclientError> {
let (payload, params) = self.take_encoded_args_and_params();
let reply = query_calculate_reply(&self.env.api, self.destination, payload, params).await;
decode_reply_or_throw::<T, _>(&self.route, reply)
}
}
impl<T: ServiceCall> Future for PendingCall<T, GclientEnv> {
type Output = Result<T::Output, <GclientEnv as GearEnv>::Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if self.state.is_none() {
let (payload, params) = self.take_encoded_args_and_params();
let send_future =
send_for_reply(self.env.api.clone(), self.destination, payload, params);
self.state = Some(Box::pin(send_future));
}
let this = self.as_mut().project();
let message_future = this
.state
.as_pin_mut()
.unwrap_or_else(|| panic!("{PENDING_CALL_INVALID_STATE}"));
let reply = ready!(message_future.poll(cx)).map(|(_, payload)| payload);
Poll::Ready(decode_reply_or_throw::<T, _>(this.route, reply))
}
}
impl<A, T> Future for PendingCtor<A, T, GclientEnv>
where
T: ServiceCall,
T::Output: PendingCtorOutput<A, GclientEnv>,
{
type Output = Result<
<T::Output as PendingCtorOutput<A, GclientEnv>>::Output,
<GclientEnv as GearEnv>::Error,
>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if self.state.is_none() {
let params = self.params.take().unwrap_or_default();
let salt = self.salt.take().unwrap();
let args = self
.args
.take()
.unwrap_or_else(|| panic!("{PENDING_CTOR_INVALID_STATE}"));
let payload = T::encode_call(&self.route, &args);
let create_program_future =
create_program(self.env.api.clone(), self.code_id, salt, payload, params);
self.state = Some(Box::pin(create_program_future));
}
let this = self.as_mut().project();
let message_future = this
.state
.as_pin_mut()
.unwrap_or_else(|| panic!("{PENDING_CTOR_INVALID_STATE}"));
let (reply, program_id) = match ready!(message_future.poll(cx)) {
Ok((program_id, payload)) => (Ok(payload), program_id),
Err(err) => (Err(err), ActorId::zero()),
};
match decode_reply_or_throw::<T, _>(this.route, reply) {
Ok(output) => Poll::Ready(Ok(output.map_result(this.env.clone(), program_id))),
Err(err) => Poll::Ready(Err(err)),
}
}
}
impl Listener for GclientEnv {
type Error = GclientError;
async fn listen<E, F: FnMut((ActorId, Vec<u8>)) -> Option<(ActorId, E)>>(
&self,
f: F,
) -> Result<impl Stream<Item = (ActorId, E)> + Unpin + use<E, F>, Self::Error> {
let listener = self.api.subscribe().await?;
let stream = stream::unfold(listener, |mut l| async move {
let vec = get_events_from_block(&mut l).await.ok();
vec.map(|v| (v, l))
})
.flat_map(stream::iter);
let stream = tokio_stream::StreamExt::filter_map(stream, f);
Ok(Box::pin(stream))
}
}
async fn create_program(
api: GearApi,
code_id: CodeId,
salt: impl AsRef<[u8]>,
payload: impl AsRef<[u8]>,
params: GclientParams,
) -> Result<(ActorId, Vec<u8>), GclientError> {
let value = params.value.unwrap_or(0);
#[cfg(not(feature = "ethexe"))]
let gas_limit = if let Some(gas_limit) = params.gas_limit {
gas_limit
} else {
let gas_info = api
.calculate_create_gas(None, code_id, Vec::from(payload.as_ref()), value, true)
.await?;
gas_info.min_limit
};
#[cfg(feature = "ethexe")]
let gas_limit = 0;
let listener = api.subscribe().await?;
let (message_id, program_id, ..) = api
.create_program_bytes(code_id, salt, payload, gas_limit, value)
.await?;
wait_for_reply_owned(listener, message_id, program_id).await
}
async fn send_for_reply(
api: GearApi,
program_id: ActorId,
payload: Vec<u8>,
params: GclientParams,
) -> Result<(ActorId, Vec<u8>), GclientError> {
let (listener, message_id) =
send_for_reply_and_listen(&api, program_id, payload, params).await?;
wait_for_reply_owned(listener, message_id, program_id).await
}
async fn send_for_reply_and_listen(
api: &GearApi,
program_id: ActorId,
payload: Vec<u8>,
params: GclientParams,
) -> Result<(EventListener, MessageId), GclientError> {
let value = params.value.unwrap_or(0);
#[cfg(not(feature = "ethexe"))]
let gas_limit = calculate_gas_limit(api, program_id, &payload, params.gas_limit, value).await?;
#[cfg(feature = "ethexe")]
let gas_limit = 0;
let listener = api.subscribe().await?;
let message_id = send_message_with_voucher_if_some(
api,
program_id,
payload,
gas_limit,
value,
params.voucher,
)
.await?;
Ok((listener, message_id))
}
async fn wait_for_reply_owned(
mut listener: EventListener,
message_id: MessageId,
program_id: ActorId,
) -> Result<(ActorId, Vec<u8>), GclientError> {
let (_, reply_code, payload, _) = wait_for_reply(&mut listener, message_id).await?;
match reply_code {
ReplyCode::Success(_) => Ok((program_id, payload)),
ReplyCode::Error(reason) => Err(GclientError::ReplyHasError(reason, payload)),
ReplyCode::Unsupported => Err(GclientError::ReplyIsMissing),
}
}
async fn send_one_way(
api: &GearApi,
program_id: ActorId,
payload: Vec<u8>,
params: GclientParams,
) -> Result<MessageId, GclientError> {
let value = params.value.unwrap_or(0);
#[cfg(not(feature = "ethexe"))]
let gas_limit = calculate_gas_limit(api, program_id, &payload, params.gas_limit, value).await?;
#[cfg(feature = "ethexe")]
let gas_limit = 0;
send_message_with_voucher_if_some(api, program_id, payload, gas_limit, value, params.voucher)
.await
}
async fn send_message_with_voucher_if_some(
api: &GearApi,
program_id: ActorId,
payload: impl AsRef<[u8]>,
gas_limit: GasUnit,
value: ValueUnit,
voucher: Option<(VoucherId, bool)>,
) -> Result<MessageId, GclientError> {
let (message_id, ..) = if let Some((voucher_id, keep_alive)) = voucher {
api.send_message_bytes_with_voucher(
voucher_id, program_id, payload, gas_limit, value, keep_alive,
)
.await?
} else {
api.send_message_bytes(program_id, payload, gas_limit, value)
.await?
};
Ok(message_id)
}
#[cfg(not(feature = "ethexe"))]
async fn calculate_gas_limit(
api: &GearApi,
program_id: ActorId,
payload: impl AsRef<[u8]>,
gas_limit: Option<GasUnit>,
value: ValueUnit,
) -> Result<u64, GclientError> {
let gas_limit = if let Some(gas_limit) = gas_limit {
gas_limit
} else {
let gas_info = api
.calculate_handle_gas(None, program_id, payload.as_ref().to_vec(), value, true)
.await?;
gas_info.min_limit
};
Ok(gas_limit)
}
async fn query_calculate_reply(
api: &GearApi,
destination: ActorId,
payload: impl AsRef<[u8]>,
params: GclientParams,
) -> Result<Vec<u8>, GclientError> {
#[cfg(not(feature = "ethexe"))]
let gas_limit = if let Some(gas_limit) = params.gas_limit {
gas_limit
} else {
api.block_gas_limit()?
};
#[cfg(feature = "ethexe")]
let gas_limit = 0;
let value = params.value.unwrap_or(0);
let origin = H256::from_slice(api.account_id().as_ref());
let payload = payload.as_ref().to_vec();
let reply_info = api
.calculate_reply_for_handle_at(
Some(origin.0.into()),
destination,
payload,
gas_limit,
value,
params.at_block.map(|at| at.0.into()),
)
.await?;
match reply_info.code {
ReplyCode::Success(_) => Ok(reply_info.payload),
ReplyCode::Error(reason) => Err(GclientError::ReplyHasError(reason, reply_info.payload)),
ReplyCode::Unsupported => Err(GclientError::ReplyIsMissing),
}
}
async fn wait_for_reply(
listener: &mut EventListener,
message_id: MessageId,
) -> Result<(MessageId, ReplyCode, Vec<u8>, ValueUnit), GclientError> {
listener
.proc(|e| {
if let ::gclient::Event::Gear(::gclient::GearEvent::UserMessageSent { message, .. }) = e
&& let Some(details) = message.details()
{
details.to_message_id().eq(&message_id).then(|| {
(
message.id(),
details.to_reply_code(),
message.payload_bytes().to_vec(),
message.value(),
)
})
} else {
None
}
})
.await
.map_err(Into::into)
}
async fn get_events_from_block(
listener: &mut EventListener,
) -> Result<Vec<(ActorId, Vec<u8>)>, GclientError> {
let vec = listener
.proc_many(
|e| {
if let ::gclient::Event::Gear(::gclient::GearEvent::UserMessageSent {
message,
..
}) = e
{
if message.destination() == ActorId::zero() {
Some((message.source(), message.payload_bytes().to_vec()))
} else {
None
}
} else {
None
}
},
|v| (v, true),
)
.await?;
Ok(vec)
}