,
/// Transactions pool
pool: Arc,
/// Subscriptions manager
subscriptions: SubscriptionManager,
/// The key store.
keystore: BareCryptoStorePtr,
/// Whether to deny unsafe calls
deny_unsafe: DenyUnsafe,
}
impl Author {
/// Create new instance of Authoring API.
pub fn new(
client: Arc,
pool: Arc,
subscriptions: SubscriptionManager,
keystore: BareCryptoStorePtr,
deny_unsafe: DenyUnsafe,
) -> Self {
Author {
client,
pool,
subscriptions,
keystore,
deny_unsafe,
}
}
}
/// Currently we treat all RPC transactions as externals.
///
/// Possibly in the future we could allow opt-in for special treatment
/// of such transactions, so that the block authors can inject
/// some unique transactions via RPC and have them included in the pool.
const TX_SOURCE: TransactionSource = TransactionSource::External;
impl AuthorApi, BlockHash> for Author
where
P: TransactionPool + Sync + Send + 'static,
Client: HeaderBackend + ProvideRuntimeApi + Send + Sync + 'static,
Client::Api: SessionKeys,
{
type Metadata = crate::metadata::Metadata;
fn insert_key(
&self,
key_type: String,
suri: String,
public: Bytes,
) -> Result<()> {
self.deny_unsafe.check_if_safe()?;
let key_type = key_type.as_str().try_into().map_err(|_| Error::BadKeyType)?;
let mut keystore = self.keystore.write();
keystore.insert_unknown(key_type, &suri, &public[..])
.map_err(|_| Error::KeyStoreUnavailable)?;
Ok(())
}
fn rotate_keys(&self) -> Result {
self.deny_unsafe.check_if_safe()?;
let best_block_hash = self.client.info().best_hash;
self.client.runtime_api().generate_session_keys(
&generic::BlockId::Hash(best_block_hash),
None,
).map(Into::into).map_err(|e| Error::Client(Box::new(e)))
}
fn has_session_keys(&self, session_keys: Bytes) -> Result {
self.deny_unsafe.check_if_safe()?;
let best_block_hash = self.client.info().best_hash;
let keys = self.client.runtime_api().decode_session_keys(
&generic::BlockId::Hash(best_block_hash),
session_keys.to_vec(),
).map_err(|e| Error::Client(Box::new(e)))?
.ok_or_else(|| Error::InvalidSessionKeys)?;
Ok(self.keystore.read().has_keys(&keys))
}
fn has_key(&self, public_key: Bytes, key_type: String) -> Result {
self.deny_unsafe.check_if_safe()?;
let key_type = key_type.as_str().try_into().map_err(|_| Error::BadKeyType)?;
Ok(self.keystore.read().has_keys(&[(public_key.to_vec(), key_type)]))
}
fn submit_extrinsic(&self, ext: Bytes) -> FutureResult> {
let xt = match Decode::decode(&mut &ext[..]) {
Ok(xt) => xt,
Err(err) => return Box::new(result(Err(err.into()))),
};
let best_block_hash = self.client.info().best_hash;
Box::new(self.pool
.submit_one(&generic::BlockId::hash(best_block_hash), TX_SOURCE, xt)
.compat()
.map_err(|e| e.into_pool_error()
.map(Into::into)
.unwrap_or_else(|e| error::Error::Verification(Box::new(e)).into()))
)
}
fn pending_extrinsics(&self) -> Result> {
Ok(self.pool.ready().map(|tx| tx.data().encode().into()).collect())
}
fn remove_extrinsic(
&self,
bytes_or_hash: Vec>>,
) -> Result>> {
self.deny_unsafe.check_if_safe()?;
let hashes = bytes_or_hash.into_iter()
.map(|x| match x {
hash::ExtrinsicOrHash::Hash(h) => Ok(h),
hash::ExtrinsicOrHash::Extrinsic(bytes) => {
let xt = Decode::decode(&mut &bytes[..])?;
Ok(self.pool.hash_of(&xt))
},
})
.collect::>>()?;
Ok(
self.pool
.remove_invalid(&hashes)
.into_iter()
.map(|tx| tx.hash().clone())
.collect()
)
}
fn watch_extrinsic(&self,
_metadata: Self::Metadata,
subscriber: Subscriber, BlockHash>>,
xt: Bytes,
) {
let submit = || -> Result<_> {
let best_block_hash = self.client.info().best_hash;
let dxt = TransactionFor::
::decode(&mut &xt[..])
.map_err(error::Error::from)?;
Ok(
self.pool
.submit_and_watch(&generic::BlockId::hash(best_block_hash), TX_SOURCE, dxt)
.map_err(|e| e.into_pool_error()
.map(error::Error::from)
.unwrap_or_else(|e| error::Error::Verification(Box::new(e)).into())
)
)
};
let subscriptions = self.subscriptions.clone();
let future = ready(submit())
.and_then(|res| res)
// convert the watcher into a `Stream`
.map(|res| res.map(|stream| stream.map(|v| Ok::<_, ()>(Ok(v)))))
// now handle the import result,
// start a new subscrition
.map(move |result| match result {
Ok(watcher) => {
subscriptions.add(subscriber, move |sink| {
sink
.sink_map_err(|_| unimplemented!())
.send_all(Compat::new(watcher))
.map(|_| ())
});
},
Err(err) => {
warn!("Failed to submit extrinsic: {}", err);
// reject the subscriber (ignore errors - we don't care if subscriber is no longer there).
let _ = subscriber.reject(err.into());
},
});
let res = self.subscriptions.executor()
.execute(Box::new(Compat::new(future.map(|_| Ok(())))));
if res.is_err() {
warn!("Error spawning subscription RPC task.");
}
}
fn unwatch_extrinsic(&self, _metadata: Option, id: SubscriptionId) -> Result {
Ok(self.subscriptions.cancel(id))
}
}