add publish_all_accounts config option

Author: Richard Patel

README.md

@@ -41,10 +41,11 @@ Config is specified via the plugin's JSON config file.
   "shutdown_timeout_ms": 30000,
   "update_account_topic": "solana.testnet.account_updates",
   "slot_status_topic": "solana.testnet.slot_status",
+  "publish_all_accounts": false,
   "program_ignores": [
     "Sysvar1111111111111111111111111111111111111",
     "Vote111111111111111111111111111111111111111"
-  ],
+  ]
 }
 ```
 
@@ -56,6 +57,7 @@ Config is specified via the plugin's JSON config file.
 - `shutdown_timeout_ms`: Time the plugin is given to flush out all messages to Kafka upon exit request.
 - `update_account_topic`: Topic name of account updates. Omit to disable.
 - `slot_status_topic`: Topic name of slot status update. Omit to disable.
+- `publish_all_accounts`: Publish all accounts on startup. Omit to disable.
 - `program_ignores`: Solana program IDs for which to ignore updates for owned accounts.
 
 ## Buffering

src/config.rs

@@ -43,6 +43,9 @@ pub struct Config {
     /// List of programs to ignore.
     #[serde(default)]
     pub program_ignores: Vec<String>,
+    /// Publish all accounts on startup.
+    #[serde(default)]
+    pub publish_all_accounts: bool,
 }
 
 impl Default for Config {
@@ -53,6 +56,7 @@ impl Default for Config {
             update_account_topic: "".to_owned(),
             slot_status_topic: "".to_owned(),
             program_ignores: Vec::new(),
+            publish_all_accounts: false,
         }
     }
 }

src/filter.rs

@@ -48,11 +48,13 @@ mod tests {
 
     #[test]
     fn test_filter() {
-        let mut config = Config::default();
-        config.program_ignores = vec![
-            "Sysvar1111111111111111111111111111111111111".to_owned(),
-            "Vote111111111111111111111111111111111111111".to_owned(),
-        ];
+        let config = Config {
+            program_ignores: vec![
+                "Sysvar1111111111111111111111111111111111111".to_owned(),
+                "Vote111111111111111111111111111111111111111".to_owned(),
+            ],
+            ..Config::default()
+        };
 
         let filter = Filter::new(&config);
         assert_eq!(filter.program_ignores.len(), 2);

src/plugin.rs

@@ -28,6 +28,7 @@ use {
 pub struct KafkaPlugin {
     publisher: Option<Publisher>,
     filter: Option<Filter>,
+    publish_all_accounts: bool,
 }
 
 impl Debug for KafkaPlugin {
@@ -54,6 +55,7 @@ impl AccountsDbPlugin for KafkaPlugin {
             config_file
         );
         let config = Config::read_from(config_file)?;
+        self.publish_all_accounts = config.publish_all_accounts;
 
         let (version_n, version_s) = get_rdkafka_version();
         info!("rd_kafka_version: {:#08x}, {}", version_n, version_s);
@@ -82,7 +84,7 @@ impl AccountsDbPlugin for KafkaPlugin {
         slot: u64,
         is_startup: bool,
     ) -> PluginResult<()> {
-        if is_startup {
+        if is_startup && !self.publish_all_accounts {
             return Ok(());
         }