| 31 | |
| 32 | impl<'a> Integration<'a> { |
| 33 | pub fn new(conf: &Config) -> Result<Integration<'a>> { |
| 34 | info!("Initializing Kafka integration"); |
| 35 | |
| 36 | // event-key template. |
| 37 | let mut templates = Handlebars::new(); |
| 38 | templates.register_escape_fn(handlebars::no_escape); |
| 39 | templates.register_template_string("event_key", &conf.event_key)?; |
| 40 | |
| 41 | let producer: FutureProducer = ClientConfig::new() |
| 42 | .set("bootstrap.servers", conf.brokers.join(",")) |
| 43 | .set("message.timeout.ms", "5000") |
| 44 | .set("allow.auto.create.topics", "true") |
| 45 | .set( |
| 46 | "sasl.mechanism", |
| 47 | match conf.mechanism.as_ref() { |
| 48 | "PLAIN" => "PLAIN", |
| 49 | "SCRAM-SHA-256" => "SCRAM-SHA-256", |
| 50 | "SCRAM-SHA-512" => "SCRAM-SHA-512", |
| 51 | _ => { |
| 52 | return Err(anyhow!( |
| 53 | "mechanism must be PLAIN, SCRAM-SHA-256 or SCRAM-SHA-512" |
| 54 | )); |
| 55 | } |
| 56 | }, |
| 57 | ) |
| 58 | .set("sasl.username", &conf.username) |
| 59 | .set("sasl.password", &conf.password) |
| 60 | .create()?; |
| 61 | |
| 62 | let i = Integration { |
| 63 | templates, |
| 64 | producer, |
| 65 | json: conf.json, |
| 66 | topic: conf.topic.clone(), |
| 67 | }; |
| 68 | |
| 69 | Ok(i) |
| 70 | } |
| 71 | |
| 72 | async fn publish_event(&self, event: &str, event_key: String, b: &[u8]) -> Result<()> { |
| 73 | info!(topic = %self.topic, event_key = %event_key, "Publishing event"); |