## ----include = FALSE---------------------------------------------------------- knitr::opts_chunk$set( collapse = TRUE, comment = "#>", eval = FALSE ) ## ----setup-------------------------------------------------------------------- # library(multichainr) # # # Set the path to MultiChain binaries # mc_set_path(Sys.getenv("MULTICHAIN_PATH")) ## ----init--------------------------------------------------------------------- # chain_name <- "streams_demo_chain" # # # Create and start the node # mc_node_init(chain_name) # mc_node_start(chain_name) # # # Wait for the node to initialize # Sys.sleep(3) # # # Connect to the local node # config <- mc_get_config(chain_name) # conn <- mc_connect(config) ## ----create_stream------------------------------------------------------------ # stream_name <- "sensor_data" # # # Create an open stream # mc_create_stream(conn, stream_name, open = TRUE) # # # Before reading from a stream, the node must be subscribed to it. # # This instructs the node to index the stream's items locally. # mc_subscribe(conn, stream_name) # # # Verify stream information # info <- mc_get_stream_info(conn, stream_name) # print(info$name) ## ----publishing--------------------------------------------------------------- # # 1. Publish a simple text message # mc_publish(conn, stream_name, "device_01", list(text = "Temperature: 22.5C")) # # # 2. Publish structured JSON data # sensor_log <- list( # temp = 23.1, # humidity = 45, # status = "OK" # ) # mc_publish(conn, stream_name, "device_01", list(json = sensor_log)) # # # 3. Publish an item with multiple keys # mc_publish(conn, stream_name, c("device_02", "alert"), list(text = "Critical Battery Level")) ## ----retrieval---------------------------------------------------------------- # # List the 10 most recent items in the stream # # Returns a data frame with columns: publishers, key, data, blocktime, etc. # recent_items <- mc_list_stream_items(conn, stream_name, count = 10) # print(recent_items) # # # List all items associated with a specific key # device_history <- mc_list_stream_key_items(conn, stream_name, "device_01") # print(device_history) ## ----summary------------------------------------------------------------------ # # Update the status of device_01 # mc_publish(conn, stream_name, "device_01", list(json = list(status = "MAINTENANCE"))) # # # Get the merged summary for 'device_01'. # # We use "jsonobjectmerge,ignoreother" to skip the plain text items # # we published earlier. # current_state <- mc_get_stream_key_summary(conn, # stream_name, # "device_01", # mode = "jsonobjectmerge,ignoreother") # print(current_state) ## ----cleanup------------------------------------------------------------------ # # Stop the node # mc_node_stop(conn) # Sys.sleep(2) # # # Determine data directory # if (.Platform$OS.type == "windows") { # base_dir <- file.path(Sys.getenv("APPDATA"), "MultiChain") # } else if (Sys.info()["sysname"] == "Darwin") { # base_dir <- file.path(Sys.getenv("HOME"), "Library/Application Support/MultiChain") # } else { # base_dir <- file.path(Sys.getenv("HOME"), ".multichain") # } # # chain_dir <- file.path(base_dir, chain_name) # if (dir.exists(chain_dir)) { # unlink(chain_dir, recursive = TRUE) # }