Events (#89)

.travis.yml

@@ -55,6 +55,7 @@ jobs:
         - pushd examples/interface && anchor test && popd
         - pushd examples/lockup && anchor test && popd
         - pushd examples/misc && anchor test && popd
+        - pushd examples/events && anchor test && popd
         - pushd examples/cashiers-check && anchor test && popd
     - <<: *examples
       name: Runs the examples 2

CHANGELOG.md

@@ -15,6 +15,11 @@ incremented for features.
 
 * cli: Specify test files to run ([#118](https://github.com/project-serum/anchor/pull/118)).
 * lang: Allow overriding the `#[state]` account's size ([#121](https://github.com/project-serum/anchor/pull/121)).
+* lang, client, ts: Add event emission and subscriptions ([#89](https://github.com/project-serum/anchor/pull/89)).
+
+## Breaking Changes
+
+* client: Replace url str with `Cluster` struct when constructing clients ([#89](https://github.com/project-serum/anchor/pull/89)).
 
 ## [0.3.0] - 2021-03-12
 

Cargo.lock

@@ -75,6 +75,17 @@ dependencies = [
  "syn 1.0.57",
 ]
 
+[[package]]
+name = "anchor-attribute-event"
+version = "0.3.0"
+dependencies = [
+ "anchor-syn",
+ "anyhow",
+ "proc-macro2 1.0.24",
+ "quote 1.0.9",
+ "syn 1.0.57",
+]
+
 [[package]]
 name = "anchor-attribute-interface"
 version = "0.3.0"
@@ -138,6 +149,8 @@ name = "anchor-client"
 version = "0.3.0"
 dependencies = [
  "anchor-lang",
+ "anyhow",
+ "regex",
  "solana-client",
  "solana-sdk",
  "thiserror",
@@ -161,10 +174,12 @@ dependencies = [
  "anchor-attribute-access-control",
  "anchor-attribute-account",
  "anchor-attribute-error",
+ "anchor-attribute-event",
  "anchor-attribute-interface",
  "anchor-attribute-program",
  "anchor-attribute-state",
  "anchor-derive-accounts",
+ "base64 0.13.0",
  "borsh",
  "solana-program",
  "thiserror",
@@ -2367,14 +2382,13 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.4.3"
+version = "1.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9251239e129e16308e70d853559389de218ac275b515068abc96829d05b948a"
+checksum = "957056ecddbeba1b26965114e191d2e8589ce74db242b6ea25fc4062427a5c19"
 dependencies = [
  "aho-corasick",
  "memchr",
  "regex-syntax",
- "thread_local",
 ]
 
 [[package]]
@@ -3487,15 +3501,6 @@ dependencies = [
  "syn 1.0.57",
 ]
 
-[[package]]
-name = "thread_local"
-version = "1.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8018d24e04c95ac8790716a5987d0fec4f8b27249ffa0f7d33f1369bdfb88cbd"
-dependencies = [
- "once_cell",
-]
-
 [[package]]
 name = "time"
 version = "0.1.43"

client/Cargo.toml

@@ -8,6 +8,8 @@ description = "Rust client for Anchor programs"
 
 [dependencies]
 anchor-lang = { path = "../lang", version = "0.3.0" }
-solana-client = "1.5.8"
-solana-sdk = "1.5.8"
+anyhow = "1.0.32"
+regex = "1.4.5"
+solana-client = "=1.5.15"
+solana-sdk = "=1.5.15"
 thiserror = "1.0.20"

client/example/Cargo.toml

@@ -7,9 +7,10 @@ edition = "2018"
 [workspace]
 
 [dependencies]
-anchor-client = { git = "https://github.com/project-serum/anchor" }
+anchor-client = { path = "../" }
 basic-2 = { path = "../../examples/tutorial/basic-2/programs/basic-2", features = ["no-entrypoint"] }
 composite = { path = "../../examples/composite/programs/composite", features = ["no-entrypoint"] }
+events = { path = "../../examples/events/programs/events", features = ["no-entrypoint"] }
 shellexpand = "2.1.0"
 anyhow = "1.0.32"
 rand = "0.7.3"

client/example/run-test.sh

@@ -38,11 +38,15 @@ main() {
     anchor deploy
     local basic_2_pid=$(cat target/idl/basic_2.json | jq -r .metadata.address)
     popd
-
+    pushd ../../examples/events
+    anchor build
+    anchor deploy
+    local events_pid=$(cat target/idl/events.json | jq -r .metadata.address)
+    popd
     #
     # Run Test.
     #
-    cargo run -- --composite-pid $composite_pid --basic-2-pid $basic_2_pid
+    cargo run -- --composite-pid $composite_pid --basic-2-pid $basic_2_pid --events-pid $events_pid
 }
 
 cleanup() {

client/example/src/main.rs

@@ -4,18 +4,21 @@ use anchor_client::solana_sdk::signature::read_keypair_file;
 use anchor_client::solana_sdk::signature::{Keypair, Signer};
 use anchor_client::solana_sdk::system_instruction;
 use anchor_client::solana_sdk::sysvar;
-use anchor_client::Client;
+use anchor_client::{Client, Cluster, EventContext};
 use anyhow::Result;
 // The `accounts` and `instructions` modules are generated by the framework.
 use basic_2::accounts as basic_2_accounts;
 use basic_2::instruction as basic_2_instruction;
 use basic_2::Counter;
+use events::instruction as events_instruction;
+use events::MyEvent;
 // The `accounts` and `instructions` modules are generated by the framework.
 use clap::Clap;
 use composite::accounts::{Bar, CompositeUpdate, Foo, Initialize};
 use composite::instruction as composite_instruction;
 use composite::{DummyA, DummyB};
 use rand::rngs::OsRng;
+use std::time::Duration;
 
 #[derive(Clap)]
 pub struct Opts {
@@ -23,6 +26,8 @@ pub struct Opts {
     composite_pid: Pubkey,
     #[clap(long)]
     basic_2_pid: Pubkey,
+    #[clap(long)]
+    events_pid: Pubkey,
 }
 
 // This example assumes a local validator is running with the programs
@@ -33,7 +38,10 @@ fn main() -> Result<()> {
     // Wallet and cluster params.
     let payer = read_keypair_file(&*shellexpand::tilde("~/.config/solana/id.json"))
         .expect("Example requires a keypair file");
-    let url = "http://localhost:8899";
+    let url = Cluster::Custom(
+        "http://localhost:8899".to_string(),
+        "ws://127.0.0.1:8900".to_string(),
+    );
 
     // Client.
     let client = Client::new_with_options(url, payer, CommitmentConfig::processed());
@@ -41,6 +49,7 @@ fn main() -> Result<()> {
     // Run tests.
     composite(&client, opts.composite_pid)?;
     basic_2(&client, opts.basic_2_pid)?;
+    events(&client, opts.events_pid)?;
 
     // Success.
     Ok(())
@@ -155,3 +164,35 @@ fn basic_2(client: &Client, pid: Pubkey) -> Result<()> {
 
     Ok(())
 }
+
+fn events(client: &Client, pid: Pubkey) -> Result<()> {
+    let program = client.program(pid);
+
+    let (sender, receiver) = std::sync::mpsc::channel();
+    let handle = program.on(move |_ctx: &EventContext, event: MyEvent| {
+        sender.send(event).unwrap();
+    })?;
+
+    std::thread::sleep(Duration::from_millis(1000));
+
+    program
+        .request()
+        .args(events_instruction::Initialize {})
+        .send()?;
+
+    let event = receiver.recv().unwrap();
+    assert_eq!(event.data, 5);
+    assert_eq!(event.label, "hello".to_string());
+
+    // TODO: remove once https://github.com/solana-labs/solana/issues/16102
+    //       is addressed. Until then, drop the subscription handle in another
+    //       thread so that we deadlock in the other thread as to not block
+    //       this thread.
+    std::thread::spawn(move || {
+        drop(handle);
+    });
+
+    println!("Success!");
+
+    Ok(())
+}

client/src/cluster.rs

@@ -0,0 +1,102 @@
+use anyhow::Result;
+use std::str::FromStr;
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum Cluster {
+    Testnet,
+    Mainnet,
+    VipMainnet,
+    Devnet,
+    Localnet,
+    Debug,
+    Custom(String, String),
+}
+
+impl Default for Cluster {
+    fn default() -> Self {
+        Cluster::Localnet
+    }
+}
+
+impl FromStr for Cluster {
+    type Err = anyhow::Error;
+    fn from_str(s: &str) -> Result<Cluster> {
+        match s.to_lowercase().as_str() {
+            "t" | "testnet" => Ok(Cluster::Testnet),
+            "m" | "mainnet" => Ok(Cluster::Mainnet),
+            "v" | "vipmainnet" => Ok(Cluster::VipMainnet),
+            "d" | "devnet" => Ok(Cluster::Devnet),
+            "l" | "localnet" => Ok(Cluster::Localnet),
+            "g" | "debug" => Ok(Cluster::Debug),
+            _ => Err(anyhow::Error::msg(
+                "Cluster must be one of [localnet, testnet, mainnet, devnet] or be an http or https url\n",
+            )),
+        }
+    }
+}
+
+impl std::fmt::Display for Cluster {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        let clust_str = match self {
+            Cluster::Testnet => "testnet",
+            Cluster::Mainnet => "mainnet",
+            Cluster::VipMainnet => "vipmainnet",
+            Cluster::Devnet => "devnet",
+            Cluster::Localnet => "localnet",
+            Cluster::Debug => "debug",
+            Cluster::Custom(url, _ws_url) => url,
+        };
+        write!(f, "{}", clust_str)
+    }
+}
+
+impl Cluster {
+    pub fn url(&self) -> &str {
+        match self {
+            Cluster::Devnet => "https://devnet.solana.com",
+            Cluster::Testnet => "https://testnet.solana.com",
+            Cluster::Mainnet => "https://api.mainnet-beta.solana.com",
+            Cluster::VipMainnet => "https://vip-api.mainnet-beta.solana.com",
+            Cluster::Localnet => "http://127.0.0.1:8899",
+            Cluster::Debug => "http://34.90.18.145:8899",
+            Cluster::Custom(url, _ws_url) => url,
+        }
+    }
+    pub fn ws_url(&self) -> &str {
+        match self {
+            Cluster::Devnet => "wss://devnet.solana.com",
+            Cluster::Testnet => "wss://testnet.solana.com",
+            Cluster::Mainnet => "wss://api.mainnet-beta.solana.com",
+            Cluster::VipMainnet => "wss://vip-api.mainnet-beta.solana.com",
+            Cluster::Localnet => "ws://127.0.0.1:9000",
+            Cluster::Debug => "ws://34.90.18.145:9000",
+            Cluster::Custom(_url, ws_url) => ws_url,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_cluster(name: &str, cluster: Cluster) {
+        assert_eq!(Cluster::from_str(name).unwrap(), cluster);
+    }
+
+    #[test]
+    fn test_cluster_parse() {
+        test_cluster("testnet", Cluster::Testnet);
+        test_cluster("mainnet", Cluster::Mainnet);
+        test_cluster("vipmainnet", Cluster::VipMainnet);
+        test_cluster("devnet", Cluster::Devnet);
+        test_cluster("localnet", Cluster::Localnet);
+        test_cluster("debug", Cluster::Debug);
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_cluster_bad_parse() {
+        let bad_url = "httq://my_custom_url.test.net";
+        Cluster::from_str(bad_url).unwrap();
+    }
+}

client/src/lib.rs

@@ -5,8 +5,12 @@ use anchor_lang::solana_program::instruction::{AccountMeta, Instruction};
 use anchor_lang::solana_program::program_error::ProgramError;
 use anchor_lang::solana_program::pubkey::Pubkey;
 use anchor_lang::{AccountDeserialize, InstructionData, ToAccountMetas};
+use regex::Regex;
 use solana_client::client_error::ClientError as SolanaClientError;
+use solana_client::pubsub_client::{PubsubClient, PubsubClientError, PubsubClientSubscription};
 use solana_client::rpc_client::RpcClient;
+use solana_client::rpc_config::{RpcTransactionLogsConfig, RpcTransactionLogsFilter};
+use solana_client::rpc_response::{Response as RpcResponse, RpcLogsResponse};
 use solana_sdk::commitment_config::CommitmentConfig;
 use solana_sdk::signature::{Keypair, Signature, Signer};
 use solana_sdk::transaction::Transaction;
@@ -14,9 +18,15 @@ use std::convert::Into;
 use thiserror::Error;
 
 pub use anchor_lang;
+pub use cluster::Cluster;
 pub use solana_client;
 pub use solana_sdk;
 
+mod cluster;
+
+/// EventHandle unsubscribes from a program event stream on drop.
+pub type EventHandle = PubsubClientSubscription<RpcResponse<RpcLogsResponse>>;
+
 /// Client defines the base configuration for building RPC clients to
 /// communitcate with Anchor programs running on a Solana cluster. It's
 /// primary use is to build a `Program` client via the `program` method.
@@ -25,20 +35,20 @@ pub struct Client {
 }
 
 impl Client {
-    pub fn new(cluster: &str, payer: Keypair) -> Self {
+    pub fn new(cluster: Cluster, payer: Keypair) -> Self {
         Self {
             cfg: Config {
-                cluster: cluster.to_string(),
+                cluster,
                 payer,
                 options: None,
             },
         }
     }
 
-    pub fn new_with_options(cluster: &str, payer: Keypair, options: CommitmentConfig) -> Self {
+    pub fn new_with_options(cluster: Cluster, payer: Keypair, options: CommitmentConfig) -> Self {
         Self {
             cfg: Config {
-                cluster: cluster.to_string(),
+                cluster,
                 payer,
                 options: Some(options),
             },
@@ -59,7 +69,7 @@ impl Client {
 
 // Internal configuration for a client.
 struct Config {
-    cluster: String,
+    cluster: Cluster,
     payer: Keypair,
     options: Option<CommitmentConfig>,
 }
@@ -79,7 +89,7 @@ impl Program {
     pub fn request(&self) -> RequestBuilder {
         RequestBuilder::new(
             self.program_id,
-            &self.cfg.cluster,
+            &self.cfg.cluster.url(),
             Keypair::from_bytes(&self.cfg.payer.to_bytes()).unwrap(),
             self.cfg.options,
         )
@@ -88,7 +98,7 @@ impl Program {
     /// Returns the account at the given address.
     pub fn account<T: AccountDeserialize>(&self, address: Pubkey) -> Result<T, ClientError> {
         let rpc_client = RpcClient::new_with_commitment(
-            self.cfg.cluster.clone(),
+            self.cfg.cluster.url().to_string(),
             self.cfg.options.unwrap_or_default(),
         );
         let account = rpc_client
@@ -101,7 +111,7 @@ impl Program {
 
     pub fn rpc(&self) -> RpcClient {
         RpcClient::new_with_commitment(
-            self.cfg.cluster.clone(),
+            self.cfg.cluster.url().to_string(),
             self.cfg.options.unwrap_or_default(),
         )
     }
@@ -109,6 +119,163 @@ impl Program {
     pub fn id(&self) -> Pubkey {
         self.program_id
     }
+
+    pub fn on<T: anchor_lang::EventData + anchor_lang::AnchorDeserialize>(
+        &self,
+        f: impl Fn(&EventContext, T) -> () + Send + 'static,
+    ) -> Result<EventHandle, ClientError> {
+        let addresses = vec![self.program_id.to_string()];
+        let filter = RpcTransactionLogsFilter::Mentions(addresses);
+        let ws_url = self.cfg.cluster.ws_url().to_string();
+        let cfg = RpcTransactionLogsConfig {
+            commitment: self.cfg.options,
+        };
+        let self_program_str = self.program_id.to_string();
+        let (client, receiver) = PubsubClient::logs_subscribe(&ws_url, filter.clone(), cfg)?;
+        std::thread::spawn(move || {
+            loop {
+                match receiver.recv() {
+                    Ok(logs) => {
+                        let ctx = EventContext {
+                            signature: logs.value.signature.parse().unwrap(),
+                            slot: logs.context.slot,
+                        };
+                        let mut logs = &logs.value.logs[..];
+                        if logs.len() > 0 {
+                            if let Ok(mut execution) = Execution::new(&mut logs) {
+                                for l in logs {
+                                    // Parse the log.
+                                    let (event, new_program, did_pop) = {
+                                        if self_program_str == execution.program() {
+                                            handle_program_log(&self_program_str, &l)
+                                                .unwrap_or_else(|e| {
+                                                    println!(
+                                                        "Unable to parse log: {}",
+                                                        e.to_string()
+                                                    );
+                                                    std::process::exit(1);
+                                                })
+                                        } else {
+                                            let (program, did_pop) =
+                                                handle_system_log(&self_program_str, &l);
+                                            (None, program, did_pop)
+                                        }
+                                    };
+                                    // Emit the event.
+                                    if let Some(e) = event {
+                                        f(&ctx, e);
+                                    }
+                                    // Switch program context on CPI.
+                                    if let Some(new_program) = new_program {
+                                        execution.push(new_program);
+                                    }
+                                    // Program returned.
+                                    if did_pop {
+                                        execution.pop();
+                                    }
+                                }
+                            }
+                        }
+                    }
+                    Err(_err) => {
+                        return;
+                    }
+                }
+            }
+        });
+        Ok(client)
+    }
+}
+
+fn handle_program_log<T: anchor_lang::EventData + anchor_lang::AnchorDeserialize>(
+    self_program_str: &str,
+    l: &str,
+) -> Result<(Option<T>, Option<String>, bool), ClientError> {
+    // Log emitted from the current program.
+    if l.starts_with("Program log:") {
+        let log = l.to_string().split_off("Program log: ".len());
+        let borsh_bytes = anchor_lang::__private::base64::decode(log)
+            .map_err(|_| ClientError::LogParseError(l.to_string()))?;
+
+        let mut slice: &[u8] = &borsh_bytes[..];
+        let disc: [u8; 8] = {
+            let mut disc = [0; 8];
+            disc.copy_from_slice(&borsh_bytes[..8]);
+            slice = &slice[8..];
+            disc
+        };
+        let mut event = None;
+        if disc == T::discriminator() {
+            let e: T = anchor_lang::AnchorDeserialize::deserialize(&mut slice)
+                .map_err(|e| ClientError::LogParseError(e.to_string()))?;
+            event = Some(e);
+        }
+        Ok((event, None, false))
+    }
+    // System log.
+    else {
+        let (program, did_pop) = handle_system_log(&self_program_str, &l);
+        Ok((None, program, did_pop))
+    }
+}
+
+fn handle_system_log(this_program_str: &str, log: &str) -> (Option<String>, bool) {
+    if log.starts_with(&format!("Program {} log:", this_program_str)) {
+        (Some(this_program_str.to_string()), false)
+    } else if log.contains("invoke") {
+        (Some("cpi".to_string()), false) // Any string will do.
+    } else {
+        let re = Regex::new(r"^Program (.*) success*$").unwrap();
+        if re.is_match(log) {
+            (None, true)
+        } else {
+            (None, false)
+        }
+    }
+}
+
+struct Execution {
+    stack: Vec<String>,
+}
+
+impl Execution {
+    pub fn new(logs: &mut &[String]) -> Result<Self, ClientError> {
+        let l = &logs[0];
+        *logs = &logs[1..];
+
+        let re = Regex::new(r"^Program (.*) invoke.*$").unwrap();
+        let c = re
+            .captures(l)
+            .ok_or(ClientError::LogParseError(l.to_string()))?;
+        let program = c
+            .get(1)
+            .ok_or(ClientError::LogParseError(l.to_string()))?
+            .as_str()
+            .to_string();
+        Ok(Self {
+            stack: vec![program],
+        })
+    }
+
+    pub fn program(&self) -> String {
+        assert!(self.stack.len() > 0);
+        self.stack[self.stack.len() - 1].clone()
+    }
+
+    pub fn push(&mut self, new_program: String) {
+        self.stack.push(new_program);
+    }
+
+    pub fn pop(&mut self) {
+        assert!(self.stack.len() > 0);
+        self.stack.pop().unwrap();
+    }
+}
+
+#[derive(Debug)]
+pub struct EventContext {
+    pub signature: Signature,
+    pub slot: u64,
 }
 
 #[derive(Debug, Error)]
@@ -119,6 +286,10 @@ pub enum ClientError {
     ProgramError(#[from] ProgramError),
     #[error("{0}")]
     SolanaClientError(#[from] SolanaClientError),
+    #[error("{0}")]
+    SolanaClientPubsubError(#[from] PubsubClientError),
+    #[error("Unable to parse log: {0}")]
+    LogParseError(String),
 }
 
 /// `RequestBuilder` provides a builder interface to create and send
@@ -225,3 +396,34 @@ impl<'a> RequestBuilder<'a> {
             .map_err(Into::into)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn new_execution() {
+        let mut logs: &[String] =
+            &["Program 7Y8VDzehoewALqJfyxZYMgYCnMTCDhWuGfJKUvjYWATw invoke [1]".to_string()];
+        let exe = Execution::new(&mut logs).unwrap();
+        assert_eq!(
+            exe.stack[0],
+            "7Y8VDzehoewALqJfyxZYMgYCnMTCDhWuGfJKUvjYWATw".to_string()
+        );
+    }
+
+    #[test]
+    fn handle_system_log_pop() {
+        let log = "Program 7Y8VDzehoewALqJfyxZYMgYCnMTCDhWuGfJKUvjYWATw success";
+        let (program, did_pop) = handle_system_log("asdf", log);
+        assert_eq!(program, None);
+        assert_eq!(did_pop, true);
+    }
+
+    #[test]
+    fn handle_system_log_no_pop() {
+        let log = "Program 7swsTUiQ6KUK4uFYquQKg4epFRsBnvbrTf2fZQCa2sTJ qwer";
+        let (program, did_pop) = handle_system_log("asdf", log);
+        assert_eq!(program, None);
+        assert_eq!(did_pop, false);
+    }
+}

examples/composite/programs/composite/Cargo.toml

@@ -13,4 +13,4 @@ no-entrypoint = []
 cpi = ["no-entrypoint"]
 
 [dependencies]
-anchor-lang = { git = "https://github.com/project-serum/anchor", features = ["derive"] }
+anchor-lang = { path = "../../../../lang" }

examples/events/Anchor.toml

@@ -0,0 +1,2 @@
+cluster = "localnet"
+wallet = "~/.config/solana/id.json"

examples/events/Cargo.toml

@@ -0,0 +1,4 @@
+[workspace]
+members = [
+    "programs/*"
+]

examples/events/migrations/deploy.js

@@ -0,0 +1,12 @@
+// Migrations are an early feature. Currently, they're nothing more than this
+// single deploy script that's invoked from the CLI, injecting a provider
+// configured from the workspace's Anchor.toml.
+
+const anchor = require("@project-serum/anchor");
+
+module.exports = async function (provider) {
+  // Configure client to use the provider.
+  anchor.setProvider(provider);
+
+  // Add your deploy script here.
+}

examples/events/programs/events/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "events"
+version = "0.1.0"
+description = "Created with Anchor"
+edition = "2018"
+
+[lib]
+crate-type = ["cdylib", "lib"]
+name = "events"
+
+[features]
+no-entrypoint = []
+no-idl = []
+cpi = ["no-entrypoint"]
+default = []
+
+[dependencies]
+anchor-lang = { path = "../../../../lang" }

examples/events/programs/events/Xargo.toml

@@ -0,0 +1,2 @@
+[target.bpfel-unknown-unknown.dependencies.std]
+features = []

examples/events/programs/events/src/lib.rs

@@ -0,0 +1,25 @@
+#![feature(proc_macro_hygiene)]
+
+use anchor_lang::prelude::*;
+
+#[program]
+pub mod events {
+    use super::*;
+    pub fn initialize(_ctx: Context<Initialize>) -> ProgramResult {
+        emit!(MyEvent {
+            data: 5,
+            label: "hello".to_string(),
+        });
+        Ok(())
+    }
+}
+
+#[derive(Accounts)]
+pub struct Initialize {}
+
+#[event]
+pub struct MyEvent {
+    pub data: u64,
+    #[index]
+    pub label: String,
+}

examples/events/tests/events.js

@@ -0,0 +1,25 @@
+const anchor = require('@project-serum/anchor');
+const assert = require("assert");
+
+describe("events", () => {
+  // Configure the client to use the local cluster.
+  anchor.setProvider(anchor.Provider.env());
+
+  it("Is initialized!", async () => {
+    const program = anchor.workspace.Events;
+
+    let listener = null;
+
+    let [event, slot] = await new Promise((resolve, _reject) => {
+      listener = program.addEventListener("MyEvent", (event, slot) => {
+        resolve([event, slot]);
+      });
+      program.rpc.initialize();
+    });
+    await program.removeEventListener(listener);
+
+    assert.ok(slot > 0);
+    assert.ok(event.data.toNumber() === 5);
+    assert.ok(event.label === "hello");
+  });
+});

examples/tutorial/basic-2/programs/basic-2/Cargo.toml

@@ -13,4 +13,4 @@ no-entrypoint = []
 cpi = ["no-entrypoint"]
 
 [dependencies]
-anchor-lang = { git = "https://github.com/project-serum/anchor", features = ["derive"] }
+anchor-lang = { path = "../../../../../lang" }

lang/Cargo.toml

@@ -19,6 +19,8 @@ anchor-attribute-program = { path = "./attribute/program", version = "0.3.0" }
 anchor-attribute-state = { path = "./attribute/state", version = "0.3.0" }
 anchor-attribute-interface = { path = "./attribute/interface", version = "0.3.0" }
 anchor-derive-accounts = { path = "./derive/accounts", version = "0.3.0" }
+anchor-attribute-event = { path = "./attribute/event", version = "0.3.0" }
 borsh = "0.8.2"
 solana-program = "=1.5.15"
 thiserror = "1.0.20"
+base64 = "0.13.0"

lang/attribute/account/src/lib.rs

@@ -66,6 +66,12 @@ pub fn account(
                     .map_err(|_| ProgramError::InvalidAccountData)
             }
         }
+
+        impl anchor_lang::Discriminator for #account_name {
+            fn discriminator() -> [u8; 8] {
+                #discriminator
+            }
+        }
     };
 
     proc_macro::TokenStream::from(quote! {

lang/attribute/event/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "anchor-attribute-event"
+version = "0.3.0"
+authors = ["Serum Foundation <foundation@projectserum.com>"]
+repository = "https://github.com/project-serum/anchor"
+license = "Apache-2.0"
+description = "Anchor attribute macro for defining an event"
+edition = "2018"
+
+[lib]
+proc-macro = true
+
+[dependencies]
+proc-macro2 = "1.0"
+quote = "1.0"
+syn = { version = "=1.0.57", features = ["full"] }
+anyhow = "1.0.32"
+anchor-syn = { path = "../../syn", version = "0.3.0", features = ["hash"] }

lang/attribute/event/src/lib.rs

@@ -0,0 +1,63 @@
+extern crate proc_macro;
+
+use quote::quote;
+use syn::parse_macro_input;
+
+/// A data structure representing a Solana account.
+#[proc_macro_attribute]
+pub fn event(
+    _args: proc_macro::TokenStream,
+    input: proc_macro::TokenStream,
+) -> proc_macro::TokenStream {
+    let event_strct = parse_macro_input!(input as syn::ItemStruct);
+
+    let event_name = &event_strct.ident;
+
+    let discriminator: proc_macro2::TokenStream = {
+        let discriminator_preimage = format!("event:{}", event_name.to_string());
+        let mut discriminator = [0u8; 8];
+        discriminator.copy_from_slice(
+            &anchor_syn::hash::hash(discriminator_preimage.as_bytes()).to_bytes()[..8],
+        );
+        format!("{:?}", discriminator).parse().unwrap()
+    };
+
+    proc_macro::TokenStream::from(quote! {
+        #[derive(anchor_lang::EventIndex, AnchorSerialize, AnchorDeserialize)]
+        #event_strct
+
+        impl anchor_lang::EventData for #event_name {
+            fn data(&self) -> Vec<u8> {
+                let mut d = #discriminator.to_vec();
+                d.append(&mut self.try_to_vec().unwrap());
+                d
+            }
+        }
+
+        impl anchor_lang::Discriminator for #event_name {
+            fn discriminator() -> [u8; 8] {
+                #discriminator
+            }
+        }
+    })
+}
+
+#[proc_macro]
+pub fn emit(input: proc_macro::TokenStream) -> proc_macro::TokenStream {
+    let data: proc_macro2::TokenStream = input.into();
+    proc_macro::TokenStream::from(quote! {
+        {
+            let data = anchor_lang::EventData::data(&#data);
+            let msg_str = &anchor_lang::__private::base64::encode(data);
+            anchor_lang::solana_program::msg!(msg_str);
+        }
+    })
+}
+
+// EventIndex is a marker macro. It functionally does nothing other than
+// allow one to mark fields with the `#[index]` inert attribute, which is
+// used to add metadata to IDLs.
+#[proc_macro_derive(EventIndex, attributes(index))]
+pub fn derive_event(_item: proc_macro::TokenStream) -> proc_macro::TokenStream {
+    proc_macro::TokenStream::from(quote! {})
+}

lang/attribute/state/src/lib.rs

@@ -26,7 +26,7 @@ pub fn state(
             // as the initialized value. Use the default implementation.
             quote! {
                 impl anchor_lang::AccountSize for #struct_ident {
-                    fn size(&self) -> Result<u64, ProgramError> {
+                    fn size(&self) -> std::result::Result<u64, anchor_lang::solana_program::program_error::ProgramError> {
                         Ok(8 + self
                            .try_to_vec()
                            .map_err(|_| ProgramError::Custom(1))?
@@ -39,7 +39,7 @@ pub fn state(
             // Size override given to the macro. Use it.
             quote! {
                 impl anchor_lang::AccountSize for #struct_ident {
-                    fn size(&self) -> Result<u64, ProgramError> {
+                    fn size(&self) -> std::result::Result<u64, anchor_lang::solana_program::program_error::ProgramError> {
                         Ok(#size)
                     }
                 }

lang/src/lib.rs

@@ -41,6 +41,11 @@ mod state;
 mod sysvar;
 mod vec;
 
+// Internal module used by macros.
+pub mod __private {
+    pub use base64;
+}
+
 pub use crate::context::{Context, CpiContext};
 pub use crate::cpi_account::CpiAccount;
 pub use crate::ctor::Ctor;
@@ -50,6 +55,7 @@ pub use crate::sysvar::Sysvar;
 pub use anchor_attribute_access_control::access_control;
 pub use anchor_attribute_account::account;
 pub use anchor_attribute_error::error;
+pub use anchor_attribute_event::{emit, event, EventIndex};
 pub use anchor_attribute_interface::interface;
 pub use anchor_attribute_program::program;
 pub use anchor_attribute_state::state;
@@ -175,11 +181,21 @@ pub trait AccountSize: AnchorSerialize {
     fn size(&self) -> Result<u64, ProgramError>;
 }
 
+/// The serialized event data to be emitted via a Solana log.
+pub trait EventData: AnchorSerialize + Discriminator {
+    fn data(&self) -> Vec<u8>;
+}
+
+/// 8 byte identifier for a type.
+pub trait Discriminator {
+    fn discriminator() -> [u8; 8];
+}
+
 /// The prelude contains all commonly used components of the crate.
 /// All programs should include it via `anchor_lang::prelude::*;`.
 pub mod prelude {
     pub use super::{
-        access_control, account, error, interface, program, state, AccountDeserialize,
+        access_control, account, emit, error, event, interface, program, state, AccountDeserialize,
         AccountSerialize, Accounts, AccountsExit, AccountsInit, AnchorDeserialize, AnchorSerialize,
         Context, CpiAccount, CpiContext, Ctor, ProgramAccount, ProgramState, Sysvar, ToAccountInfo,
         ToAccountInfos, ToAccountMetas,

lang/syn/src/idl.rs

@@ -12,6 +12,8 @@ pub struct Idl {
     #[serde(skip_serializing_if = "Vec::is_empty", default)]
     pub types: Vec<IdlTypeDef>,
     #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub events: Option<Vec<IdlEvent>>,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
     pub errors: Option<Vec<IdlErrorCode>>,
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub metadata: Option<serde_json::Value>,
@@ -64,6 +66,20 @@ pub struct IdlField {
     pub ty: IdlType,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct IdlEvent {
+    pub name: String,
+    pub fields: Vec<IdlEventField>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct IdlEventField {
+    pub name: String,
+    #[serde(rename = "type")]
+    pub ty: IdlType,
+    pub index: bool,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct IdlTypeDef {
     pub name: String,

lang/syn/src/parser/file.rs

@@ -176,6 +176,36 @@ pub fn parse(filename: impl AsRef<Path>) -> Result<Idl> {
         })
         .collect::<Vec<_>>();
 
+    let events = parse_events(&f)
+        .iter()
+        .map(|e: &&syn::ItemStruct| {
+            let fields = match &e.fields {
+                syn::Fields::Named(n) => n,
+                _ => panic!("Event fields must be named"),
+            };
+            let fields = fields
+                .named
+                .iter()
+                .map(|f: &syn::Field| {
+                    let index = match f.attrs.iter().next() {
+                        None => false,
+                        Some(i) => parser::tts_to_string(&i.path) == "index",
+                    };
+                    IdlEventField {
+                        name: f.ident.clone().unwrap().to_string(),
+                        ty: parser::tts_to_string(&f.ty).to_string().parse().unwrap(),
+                        index,
+                    }
+                })
+                .collect::<Vec<IdlEventField>>();
+
+            IdlEvent {
+                name: e.ident.to_string(),
+                fields,
+            }
+        })
+        .collect::<Vec<IdlEvent>>();
+
     // All user defined types.
     let mut accounts = vec![];
     let mut types = vec![];
@@ -188,7 +218,7 @@ pub fn parse(filename: impl AsRef<Path>) -> Result<Idl> {
         if ty_def.name != error_name {
             if acc_names.contains(&ty_def.name) {
                 accounts.push(ty_def);
-            } else {
+            } else if events.iter().position(|e| e.name == ty_def.name).is_none() {
                 types.push(ty_def);
             }
         }
@@ -201,6 +231,11 @@ pub fn parse(filename: impl AsRef<Path>) -> Result<Idl> {
         instructions,
         types,
         accounts,
+        events: if events.is_empty() {
+            None
+        } else {
+            Some(events)
+        },
         errors: error_codes,
         metadata: None,
     })
@@ -256,6 +291,31 @@ fn parse_error_enum(f: &syn::File) -> Option<syn::ItemEnum> {
         .next()
         .cloned()
 }
+
+fn parse_events(f: &syn::File) -> Vec<&syn::ItemStruct> {
+    f.items
+        .iter()
+        .filter_map(|i| match i {
+            syn::Item::Struct(item_strct) => {
+                let attrs_count = item_strct
+                    .attrs
+                    .iter()
+                    .filter(|attr| {
+                        let segment = attr.path.segments.last().unwrap();
+                        segment.ident == "event"
+                    })
+                    .count();
+                match attrs_count {
+                    0 => None,
+                    1 => Some(item_strct),
+                    _ => panic!("Invalid syntax: one event attribute allowed"),
+                }
+            }
+            _ => None,
+        })
+        .collect()
+}
+
 // Parse all structs implementing the `Accounts` trait.
 fn parse_accounts(f: &syn::File) -> HashMap<String, AccountsStruct> {
     f.items

ts/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@project-serum/anchor",
-  "version": "0.3.0",
+  "version": "0.3.1-beta.2",
   "description": "Anchor client",
   "main": "dist/cjs/index.js",
   "module": "dist/esm/index.js",
@@ -24,11 +24,12 @@
   },
   "dependencies": {
     "@project-serum/borsh": "^0.1.0",
-    "@solana/web3.js": "^0.90.4",
+    "@solana/web3.js": "^1.1.0",
     "@types/bn.js": "^4.11.6",
     "@types/bs58": "^4.0.1",
     "@types/crypto-hash": "^1.1.2",
     "@types/pako": "^1.0.1",
+    "base64-js": "^1.5.1",
     "bn.js": "^5.1.2",
     "bs58": "^4.0.1",
     "buffer-layout": "^1.2.0",

ts/src/coder.ts

@@ -51,10 +51,16 @@ export default class Coder {
    */
   readonly state: StateCoder;
 
+  /**
+   * Coder for events.
+   */
+  readonly events: EventCoder;
+
   constructor(idl: Idl) {
     this.instruction = new InstructionCoder(idl);
     this.accounts = new AccountsCoder(idl);
     this.types = new TypesCoder(idl);
+    this.events = new EventCoder(idl);
     if (idl.state) {
       this.state = new StateCoder(idl);
     }
@@ -201,6 +207,46 @@ class TypesCoder {
   }
 }
 
+class EventCoder {
+  /**
+   * Maps account type identifier to a layout.
+   */
+  private layouts: Map<string, Layout>;
+
+  public constructor(idl: Idl) {
+    if (idl.events === undefined) {
+      this.layouts = new Map();
+      return;
+    }
+    const layouts = idl.events.map((event) => {
+      let eventTypeDef: IdlTypeDef = {
+        name: event.name,
+        type: {
+          kind: "struct",
+          fields: event.fields.map((f) => {
+            return { name: f.name, type: f.type };
+          }),
+        },
+      };
+      return [event.name, IdlCoder.typeDefLayout(eventTypeDef, idl.types)];
+    });
+    // @ts-ignore
+    this.layouts = new Map(layouts);
+  }
+
+  public encode<T = any>(eventName: string, account: T): Buffer {
+    const buffer = Buffer.alloc(1000); // TODO: use a tighter buffer.
+    const layout = this.layouts.get(eventName);
+    const len = layout.encode(account, buffer);
+    return buffer.slice(0, len);
+  }
+
+  public decode<T = any>(eventName: string, ix: Buffer): T {
+    const layout = this.layouts.get(eventName);
+    return layout.decode(ix);
+  }
+}
+
 class StateCoder {
   private layout: Layout;
 
@@ -364,6 +410,11 @@ export async function stateDiscriminator(name: string): Promise<Buffer> {
   return Buffer.from(sha256.digest(`account:${name}`)).slice(0, 8);
 }
 
+export function eventDiscriminator(name: string): Buffer {
+  // @ts-ignore
+  return Buffer.from(sha256.digest(`event:${name}`)).slice(0, 8);
+}
+
 // Returns the size of the type in bytes. For variable length types, just return
 // 1. Users should override this value in such cases.
 function typeSize(idl: Idl, ty: IdlType): number {

ts/src/idl.ts

@@ -8,9 +8,21 @@ export type Idl = {
   state?: IdlState;
   accounts?: IdlTypeDef[];
   types?: IdlTypeDef[];
+  events?: IdlEvent[];
   errors?: IdlErrorCode[];
 };
 
+export type IdlEvent = {
+  name: string;
+  fields: IdlEventField[];
+};
+
+export type IdlEventField = {
+  name: string;
+  type: IdlType;
+  index: boolean;
+};
+
 export type IdlInstruction = {
   name: string;
   accounts: IdlAccountItem[];

ts/src/program.ts

@@ -3,9 +3,11 @@ import { inflate } from "pako";
 import Provider from "./provider";
 import { RpcFactory } from "./rpc";
 import { Idl, idlAddress, decodeIdlAccount } from "./idl";
-import Coder from "./coder";
+import Coder, { eventDiscriminator } from "./coder";
 import { Rpcs, Ixs, Txs, Accounts, State } from "./rpc";
 import { getProvider } from "./";
+import * as base64 from "base64-js";
+import * as assert from "assert";
 
 /**
  * Program is the IDL deserialized representation of a Solana program.
@@ -100,6 +102,162 @@ export class Program {
     const inflatedIdl = inflate(idlAccount.data);
     return JSON.parse(decodeUtf8(inflatedIdl));
   }
+
+  /**
+   * Invokes the given callback everytime the given event is emitted.
+   */
+  public addEventListener<T>(
+    eventName: string,
+    callback: (event: T, slot: number) => void
+  ): Promise<void> {
+    // Values shared across log handlers.
+    const thisProgramStr = this.programId.toString();
+    const discriminator = eventDiscriminator(eventName);
+    const logStartIndex = "Program log: ".length;
+
+    // Handles logs when the current program being executing is *not* this.
+    const handleSystemLog = (log: string): [string | null, boolean] => {
+      // System component.
+      const logStart = log.split(":")[0];
+      // Recursive call.
+      if (logStart.startsWith(`Program ${this.programId.toString()} invoke`)) {
+        return [this.programId.toString(), false];
+      }
+      // Cpi call.
+      else if (logStart.includes("invoke")) {
+        return ["cpi", false]; // Any string will do.
+      } else {
+        // Did the program finish executing?
+        if (logStart.match(/^Program (.*) consumed .*$/g) !== null) {
+          return [null, true];
+        }
+        return [null, false];
+      }
+    };
+
+    // Handles logs from *this* program.
+    const handleProgramLog = (
+      log: string
+    ): [T | null, string | null, boolean] => {
+      // This is a `msg!` log.
+      if (log.startsWith("Program log:")) {
+        const logStr = log.slice(logStartIndex);
+        const logArr = Buffer.from(base64.toByteArray(logStr));
+        const disc = logArr.slice(0, 8);
+        // Only deserialize if the discriminator implies a proper event.
+        let event = null;
+        if (disc.equals(discriminator)) {
+          event = this.coder.events.decode(eventName, logArr.slice(8));
+        }
+        return [event, null, false];
+      }
+      // System log.
+      else {
+        return [null, ...handleSystemLog(log)];
+      }
+    };
+
+    // Main log handler. Returns a three element array of the event, the
+    // next program that was invoked for CPI, and a boolean indicating if
+    // a program has completed execution (and thus should be popped off the
+    // execution stack).
+    const handleLog = (
+      execution: ExecutionContext,
+      log: string
+    ): [T | null, string | null, boolean] => {
+      // Executing program is this program.
+      if (execution.program() === thisProgramStr) {
+        return handleProgramLog(log);
+      }
+      // Executing program is not this program.
+      else {
+        return [null, ...handleSystemLog(log)];
+      }
+    };
+
+    // Each log given, represents an array of messages emitted by
+    // a single transaction, which can execute many different programs across
+    // CPI boundaries. However, the subscription is only interested in the
+    // events emitted by *this* program. In achieving this, we keep track of the
+    // program execution context by parsing each log and looking for a CPI
+    // `invoke` call. If one exists, we know a new program is executing. So we
+    // push the programId onto a stack and switch the program context. This
+    // allows us to track, for a given log, which program was executing during
+    // its emission, thereby allowing us to know if a given log event was
+    // emitted by *this* program. If it was, then we parse the raw string and
+    // emit the event if the string matches the event being subscribed to.
+    //
+    // @ts-ignore
+    return this.provider.connection.onLogs(this.programId, (logs, ctx) => {
+      if (logs.err) {
+        console.error(logs);
+        return;
+      }
+
+      const logScanner = new LogScanner(logs.logs);
+      const execution = new ExecutionContext(logScanner.next() as string);
+
+      let log = logScanner.next();
+      while (log !== null) {
+        let [event, newProgram, didPop] = handleLog(execution, log);
+        if (event) {
+          callback(event, ctx.slot);
+        }
+        if (newProgram) {
+          execution.push(newProgram);
+        }
+        if (didPop) {
+          execution.pop();
+        }
+        log = logScanner.next();
+      }
+    });
+  }
+
+  public async removeEventListener(listener: number): Promise<void> {
+    // @ts-ignore
+    return this.provider.connection.removeOnLogsListener(listener);
+  }
+}
+
+// Stack frame execution context, allowing one to track what program is
+// executing for a given log.
+class ExecutionContext {
+  stack: string[];
+
+  constructor(log: string) {
+    // Assumes the first log in every transaction is an `invoke` log from the
+    // runtime.
+    const program = /^Program (.*) invoke.*$/g.exec(log)[1];
+    this.stack = [program];
+  }
+
+  program(): string {
+    assert.ok(this.stack.length > 0);
+    return this.stack[this.stack.length - 1];
+  }
+
+  push(newProgram: string) {
+    this.stack.push(newProgram);
+  }
+
+  pop() {
+    assert.ok(this.stack.length > 0);
+    this.stack.pop();
+  }
+}
+
+class LogScanner {
+  constructor(public logs: string[]) {}
+
+  next(): string | null {
+    if (this.logs.length === 0) {
+      return null;
+    }
+    let l = this.logs[0];
+    this.logs = this.logs.slice(1);
+    return l;
+  }
 }
 
 function decodeUtf8(array: Uint8Array): string {

ts/src/utils.ts

@@ -1,9 +1,8 @@
 import * as bs58 from "bs58";
 import { sha256 } from "crypto-hash";
-import { struct } from "superstruct";
 import assert from "assert";
 import { PublicKey, AccountInfo, Connection } from "@solana/web3.js";
-import { idlAddress } from './idl';
+import { idlAddress } from "./idl";
 
 export const TOKEN_PROGRAM_ID = new PublicKey(
   "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
@@ -16,9 +15,7 @@ async function getMultipleAccounts(
   Array<null | { publicKey: PublicKey; account: AccountInfo<Buffer> }>
 > {
   const args = [publicKeys.map((k) => k.toBase58()), { commitment: "recent" }];
-  // @ts-ignore
-  const unsafeRes = await connection._rpcRequest("getMultipleAccounts", args);
-  const res = GetMultipleAccountsAndContextRpcResult(unsafeRes);
+  const res = await connection._rpcRequest("getMultipleAccounts", args);
   if (res.error) {
     throw new Error(
       "failed to get info about accounts " +
@@ -71,44 +68,6 @@ async function getMultipleAccounts(
   });
 }
 
-function jsonRpcResult(resultDescription: any) {
-  const jsonRpcVersion = struct.literal("2.0");
-  return struct.union([
-    struct({
-      jsonrpc: jsonRpcVersion,
-      id: "string",
-      error: "any",
-    }),
-    struct({
-      jsonrpc: jsonRpcVersion,
-      id: "string",
-      error: "null?",
-      result: resultDescription,
-    }),
-  ]);
-}
-
-function jsonRpcResultAndContext(resultDescription: any) {
-  return jsonRpcResult({
-    context: struct({
-      slot: "number",
-    }),
-    value: resultDescription,
-  });
-}
-
-const AccountInfoResult = struct({
-  executable: "boolean",
-  owner: "string",
-  lamports: "number",
-  data: "any",
-  rentEpoch: "number?",
-});
-
-const GetMultipleAccountsAndContextRpcResult = jsonRpcResultAndContext(
-  struct.array([struct.union(["null", AccountInfoResult])])
-);
-
 const utils = {
   bs58,
   sha256,

ts/tsconfig.json

@@ -1,27 +1,27 @@
 {
   "include": ["src"],
-	"compilerOptions": {
-		"moduleResolution": "node",
+  "compilerOptions": {
+    "moduleResolution": "node",
     "module": "es6",
     "target": "es2019",
 
     "outDir": "dist/esm/",
     "rootDir": "./src",
-		"declarationDir": "dist",
+    "declarationDir": "dist",
 
     "sourceMap": true,
     "declaration": true,
     "allowSyntheticDefaultImports": true,
     "experimentalDecorators": true,
     "emitDecoratorMetadata": true,
-    "noImplicitAny": true,
+    "noImplicitAny": false,
 
     "esModuleInterop": true,
     "composite": true,
-		"baseUrl": ".",
-		"typeRoots": ["types/", "node_modules/@types"],
+    "baseUrl": ".",
+    "typeRoots": ["types/", "node_modules/@types"],
     "paths": {
-      "@solana/web3.js":  ["../../node_modules/@solana/web3.js/lib"]
+      "@solana/web3.js":  ["./node_modules/@solana/web3.js/lib"]
     }
   }
 }