Add Raw Event To Handler Callback (#147)

Author: CanardMandarin

README.md

@@ -56,8 +56,8 @@ pub struct Opts {
 #[derive(Debug)]
 pub struct Logger;
 
-impl<V: std::fmt::Debug + Sync> yellowstone_vixen::Handler<V> for Logger {
-    async fn handle(&self, value: &V) -> yellowstone_vixen::HandlerResult<()> {
+impl<V: std::fmt::Debug + Sync, R: Sync> vixen::Handler<V, R> for Logger {
+    async fn handle(&self, _value: &V, _raw: &R) -> vixen::HandlerResult<()> {
         tracing::info!(?value);
         Ok(())
     }
@@ -73,7 +73,7 @@ fn main() {
     let config = std::fs::read_to_string(config).expect("Error reading config file");
     let config = toml::from_str(&config).expect("Error parsing config");
 
-    yellowstone_vixen::Runtime<YellowstoneGrpcSourc>::builder()
+    yellowstone_vixen::Runtime<YellowstoneGrpcSource>::builder()
         .account(Pipeline::new(AccountParser, [Logger]))
         .instruction(Pipeline::new(InstructionParser, [Logger]))
         .build(config)

crates/runtime/src/filter_pipeline.rs

@@ -75,7 +75,7 @@ where
     P::Input: Send + Sync,
     H: Debug + Sync,
     for<'i> &'i H: IntoIterator,
-    for<'i> <&'i H as IntoIterator>::Item: Handler<P::Output>,
+    for<'i> <&'i H as IntoIterator>::Item: Handler<P::Output, P::Input>,
 {
     /// Handle fn for `FilterPipeline`
     ///
@@ -92,7 +92,7 @@ where
         let errs = self
             .handlers
             .into_iter()
-            .map(|h| async move { h.handle(parsed).await })
+            .map(|h| async move { h.handle(parsed, value).await })
             .collect::<futures_util::stream::FuturesUnordered<_>>()
             .filter_map(|r| async move { r.err() })
             .collect::<SmallVec<[_; 1]>>()
@@ -113,7 +113,7 @@ where
     P::Output: Send + Sync,
     H: Debug + Sync,
     for<'i> &'i H: IntoIterator,
-    for<'i> <&'i H as IntoIterator>::Item: Handler<P::Output> + Send,
+    for<'i> <&'i H as IntoIterator>::Item: Handler<P::Output, P::Input> + Send,
 {
     fn handle<'h>(
         &'h self,

crates/runtime/src/handler.rs

@@ -19,16 +19,20 @@ type BoxedError = Box<dyn std::error::Error + Send + Sync + 'static>;
 /// The result returned by a handler.
 pub type HandlerResult<T> = Result<T, BoxedError>;
 
-/// A handler callback for a parsed value.
-pub trait Handler<T> {
-    /// Consume the parsed value.
-    fn handle(&self, value: &T) -> impl Future<Output = HandlerResult<()>> + Send;
+/// A handler callback for a parsed value and its corresponding raw event.
+pub trait Handler<T, R>
+where R: Sync
+{
+    /// Consume the parsed value together with the raw event.
+    fn handle(&self, value: &T, raw_event: &R) -> impl Future<Output = HandlerResult<()>> + Send;
 }
 
-impl<T: Handler<U>, U> Handler<U> for &T {
+impl<T: Handler<U, R>, U, R> Handler<U, R> for &T
+where R: Sync
+{
     #[inline]
-    fn handle(&self, value: &U) -> impl Future<Output = HandlerResult<()>> + Send {
-        <T as Handler<U>>::handle(self, value)
+    fn handle(&self, value: &U, raw_event: &R) -> impl Future<Output = HandlerResult<()>> + Send {
+        <T as Handler<U, R>>::handle(self, value, raw_event)
     }
 }
 
@@ -145,7 +149,8 @@ impl<P, I> Pipeline<P, I>
 where
     for<'i> &'i I: IntoIterator,
     P: Parser,
-    for<'i> <&'i I as IntoIterator>::Item: Handler<P::Output>,
+    P::Input: Sync,
+    for<'i> <&'i I as IntoIterator>::Item: Handler<P::Output, P::Input>,
 {
     /// Handle fn for `Pipeline`
     ///
@@ -167,7 +172,7 @@ where
         let errs = (&self.1)
             .into_iter()
             .map(|h| async move {
-                h.handle(parsed)
+                h.handle(parsed, value)
                     .instrument(tracing::info_span!("vixen.handle",))
                     .await
             })
@@ -209,7 +214,7 @@ where
     for<'i> &'i I: IntoIterator,
     P::Input: Sync,
     P::Output: Send + Sync,
-    for<'i> <&'i I as IntoIterator>::Item: Handler<P::Output> + Send,
+    for<'i> <&'i I as IntoIterator>::Item: Handler<P::Output, P::Input> + Send,
 {
     fn handle<'h>(
         &'h self,

crates/stream/src/grpc.rs

@@ -32,8 +32,8 @@ pub enum Error {
 #[derive(Debug)]
 pub struct GrpcHandler(pub(super) broadcast::Sender<Any>);
 
-impl<T: Message + Name + Sync> Handler<T> for GrpcHandler {
-    async fn handle(&self, value: &T) -> HandlerResult<()> {
+impl<T: Message + Name + Sync, R: Sync> Handler<T, R> for GrpcHandler {
+    async fn handle(&self, value: &T, _raw: &R) -> HandlerResult<()> {
         self.0.send(Any::from_msg(value)?).ok();
         Ok(())
     }

examples/filtered-pipeline/src/main.rs

@@ -32,8 +32,8 @@ pub struct Opts {
 #[derive(Debug)]
 pub struct Logger;
 
-impl<V: std::fmt::Debug + Sync> vixen::Handler<V> for Logger {
-    async fn handle(&self, _value: &V) -> vixen::HandlerResult<()> { Ok(()) }
+impl<V: std::fmt::Debug + Sync, R: Sync> vixen::Handler<V, R> for Logger {
+    async fn handle(&self, _value: &V, _raw: &R) -> vixen::HandlerResult<()> { Ok(()) }
 }
 fn main() {
     let Opts { config } = Opts::parse();

examples/prometheus/src/main.rs

@@ -24,8 +24,8 @@ pub struct Opts {
 #[derive(Debug)]
 pub struct Logger;
 
-impl<V: std::fmt::Debug + Sync> yellowstone_vixen::Handler<V> for Logger {
-    async fn handle(&self, _value: &V) -> yellowstone_vixen::HandlerResult<()> { Ok(()) }
+impl<V: std::fmt::Debug + Sync, R: Sync> yellowstone_vixen::Handler<V, R> for Logger {
+    async fn handle(&self, _value: &V, _raw: &R) -> yellowstone_vixen::HandlerResult<()> { Ok(()) }
 }
 
 #[tokio::main]

examples/tracing/src/main.rs

@@ -26,8 +26,8 @@ pub struct Opts {
 #[derive(Debug)]
 pub struct Logger;
 
-impl<V: std::fmt::Debug + Sync> Handler<V> for Logger {
-    async fn handle(&self, _value: &V) -> HandlerResult<()> { Ok(()) }
+impl<V: std::fmt::Debug + Sync, R: Sync> Handler<V, R> for Logger {
+    async fn handle(&self, _value: &V, _raw: &R) -> HandlerResult<()> { Ok(()) }
 }
 
 #[rustfmt::skip]