graphile_worker_lifecycle_hooks 0.3.3

Lifecycle hooks for graphile_worker, a high performance Rust/PostgreSQL job queue
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

use futures::future::BoxFuture;

use crate::context::{
    AfterJobRunContext, BeforeJobRunContext, BeforeJobScheduleContext, JobRecoveryContext,
};
use crate::event::{Event, HookOutput, Interceptable};
use crate::result::{HookResult, JobRecoveryResult, JobScheduleResult};
use crate::TypeErasedHooks;

macro_rules! define_interceptor_event {
    ($event:ident, $context:ty) => {
        define_interceptor_event!($event, $context, HookResult);
    };

    ($event:ident, $context:ty, $output:ty) => {
        pub struct $event;

        impl Event for $event {
            type Context = $context;
            type Output = $output;

            fn register_boxed(
                hooks: &mut TypeErasedHooks,
                handler: Box<
                    dyn Fn(Self::Context) -> BoxFuture<'static, Self::Output> + Send + Sync,
                >,
            ) {
                hooks.get_handlers_mut::<Self>().push(handler);
            }
        }

        impl Interceptable for $context {
            type Output = $output;

            fn intercept_with(self, hooks: &TypeErasedHooks) -> BoxFuture<'_, Self::Output> {
                Box::pin(async move {
                    let Some(handlers) = hooks.get_handlers::<$event>() else {
                        return <$output as Default>::default();
                    };
                    for handler in handlers {
                        let result = handler(self.clone()).await;
                        if result.is_terminal() {
                            return result;
                        }
                    }
                    <$output as Default>::default()
                })
            }
        }
    };

    ($event:ident, $context:ty, $output:ty, $chain_field:ident) => {
        pub struct $event;

        impl Event for $event {
            type Context = $context;
            type Output = $output;

            fn register_boxed(
                hooks: &mut TypeErasedHooks,
                handler: Box<
                    dyn Fn(Self::Context) -> BoxFuture<'static, Self::Output> + Send + Sync,
                >,
            ) {
                hooks.get_handlers_mut::<Self>().push(handler);
            }
        }

        impl Interceptable for $context {
            type Output = $output;

            fn intercept_with(self, hooks: &TypeErasedHooks) -> BoxFuture<'_, Self::Output> {
                Box::pin(async move {
                    let Some(handlers) = hooks.get_handlers::<$event>() else {
                        return <$output as HookOutput>::default_with_value(self.$chain_field);
                    };

                    let mut current_value = self.$chain_field.clone();
                    for handler in handlers {
                        let mut ctx = self.clone();
                        ctx.$chain_field = current_value;
                        let result = handler(ctx).await;
                        if result.is_terminal() {
                            return result;
                        }
                        current_value = result.chain_value().unwrap();
                    }
                    <$output as HookOutput>::default_with_value(current_value)
                })
            }
        }
    };
}

define_interceptor_event!(BeforeJobRun, BeforeJobRunContext);
define_interceptor_event!(AfterJobRun, AfterJobRunContext);
define_interceptor_event!(
    BeforeJobSchedule,
    BeforeJobScheduleContext,
    JobScheduleResult,
    payload
);
define_interceptor_event!(JobRecovery, JobRecoveryContext, JobRecoveryResult);