runtime/wasm (ipfs_map): add ability to pass extra userData through

This changes the signature of ipfs.map to `ipfs.map(hash, callback,
userData, flags)`, with the corresponding change to the callback's
signature as `callback(jsonData, userData)`

Author: lutter

runtime/wasm/src/host_exports.rs

@@ -4,6 +4,7 @@ use ethabi::Token;
 use futures::sync::oneshot;
 use graph::components::ethereum::*;
 use graph::components::store::EntityKey;
+use graph::data::store;
 use graph::prelude::*;
 use graph::serde_json;
 use graph::web3::types::H160;
@@ -347,12 +348,15 @@ where
     // return value contains all entity operations that were produced by the
     // callback invocations. Each invocation of `callback` happens in its own
     // instance of a WASM module, which is identical to `module` when it was
-    // first started.
+    // first started. The signature of the callback must be
+    // `callback(JSONValue, Value)`, and the `userData` parameter is passed
+    // to the callback without any changes
     pub(crate) fn ipfs_map(
         &self,
         module: &WasmiModule<E, L, S, U>,
         link: String,
         callback: &str,
+        user_data: store::Value,
         flags: Vec<String>,
     ) -> Result<Vec<EntityOperation>, HostExportError<impl ExportError>> {
         const JSON_FLAG: &str = "json";
@@ -382,7 +386,8 @@ where
                                 valid_module.clone(),
                                 ctx.clone(),
                             )?;
-                            let result = module.handle_json_callback(&*callback, &sv.value);
+                            let result =
+                                module.handle_json_callback(&*callback, &sv.value, &user_data);
                             // Log progress every 15s
                             if last_log.elapsed() > Duration::from_secs(15) {
                                 debug!(

runtime/wasm/src/module/mod.rs

@@ -12,6 +12,7 @@ use wasmi::{
 use crate::host_exports::{self, HostExportError, HostExports};
 use crate::EventHandlerContext;
 use graph::components::ethereum::*;
+use graph::data::store;
 use graph::data::subgraph::DataSource;
 use graph::ethabi::LogParam;
 use graph::prelude::{Error as FailureError, *};
@@ -276,14 +277,16 @@ where
         mut self,
         handler_name: &str,
         value: &graph::serde_json::Value,
+        user_data: &store::Value,
     ) -> Result<Vec<EntityOperation>, FailureError> {
         let value = RuntimeValue::from(self.asc_new(value));
+        let user_data = RuntimeValue::from(self.asc_new(user_data));
 
         // Invoke the callback
-        let result = self
-            .module
-            .clone()
-            .invoke_export(handler_name, &[value], &mut self);
+        let result =
+            self.module
+                .clone()
+                .invoke_export(handler_name, &[value, user_data], &mut self);
 
         // Return either the collected entity operations or an error
         result.map(|_| self.ctx.entity_operations).map_err(|e| {
@@ -540,27 +543,31 @@ where
         &mut self,
         link_ptr: AscPtr<AscString>,
         callback: AscPtr<AscString>,
+        user_data: AscPtr<AscEnum<StoreValueKind>>,
         flags: AscPtr<Array<AscPtr<AscString>>>,
     ) -> Result<Option<RuntimeValue>, Trap> {
         let link: String = self.asc_get(link_ptr);
         let callback: String = self.asc_get(callback);
+        let user_data: store::Value = self.asc_get(user_data);
+
         let flags = self.asc_get(flags);
         let start_time = Instant::now();
-        let result = match self
-            .host_exports()
-            .ipfs_map(&self, link.clone(), &*callback, flags)
-        {
-            Ok(ops) => {
-                debug!(self.logger, "Successfully processed file with ipfs.map";
+        let result =
+            match self
+                .host_exports()
+                .ipfs_map(&self, link.clone(), &*callback, user_data, flags)
+            {
+                Ok(ops) => {
+                    debug!(self.logger, "Successfully processed file with ipfs.map";
                                      "link" => &link,
                                      "callback" => &*callback,
                                      "entity_operations" => ops.len(),
                                      "time" => start_time.elapsed().as_millis());
-                self.ctx.entity_operations.extend(ops);
-                Ok(None)
-            }
-            Err(e) => Err(e.into()),
-        };
+                    self.ctx.entity_operations.extend(ops);
+                    Ok(None)
+                }
+                Err(e) => Err(e.into()),
+            };
 
         // Advance this module's start time by the time it took to run the entire
         // ipfs_map. This has the effect of not charging this module for the time
@@ -859,6 +866,7 @@ where
                 args.nth_checked(0)?,
                 args.nth_checked(1)?,
                 args.nth_checked(2)?,
+                args.nth_checked(3)?,
             ),
             _ => panic!("Unimplemented function at {}", index),
         }

runtime/wasm/src/module/test.rs

@@ -294,18 +294,22 @@ fn ipfs_cat() {
     assert_eq!(data, "42");
 }
 
+// The user_data value we use with calls to ipfs_map
+const USER_DATA: &str = "user_data";
+
 fn make_thing(id: &str, value: &str) -> (String, EntityOperation) {
     let mut data = Entity::new();
     data.set("id", id);
     data.set("value", value);
+    data.set("extra", USER_DATA);
     let subgraph_id = SubgraphDeploymentId::new("wasmModuleTest").unwrap();
     let key = EntityKey {
         subgraph_id,
         entity_type: "Thing".to_string(),
         entity_id: id.to_string(),
     };
     (
-        format!("{{ \"id\": \"{}\", \"value\": \"{}\" }}", id, value),
+        format!("{{ \"id\": \"{}\", \"value\": \"{}\"}}", id, value),
         EntityOperation::Set { key, data },
     )
 }
@@ -329,10 +333,10 @@ fn ipfs_map() {
                 .unwrap()
                 .hash
         };
-
+        let user_data = RuntimeValue::from(module.asc_new(USER_DATA));
         let converted = module.module.clone().invoke_export(
             "ipfsMap",
-            &[RuntimeValue::from(module.asc_new(&hash))],
+            &[RuntimeValue::from(module.asc_new(&hash)), user_data],
             &mut module,
         )?;
         assert_eq!(None, converted);

runtime/wasm/wasm_test/ipfs_map.ts

@@ -121,10 +121,10 @@ class JSONValue {
  * Actual setup for the test
  */
 declare namespace ipfs {
-  function map(hash: String, callback: String, flags: String[]): void
+  function map(hash: String, callback: String, userData: Value, flags: String[]): void
 }
 
-export function echoToStore(data: JSONValue): void {
+export function echoToStore(data: JSONValue, userData: Value): void {
   // expect a map of the form { "id": "anId", "value": "aValue" }
   let map = data.toObject();
   let id = map.get("id").toString();
@@ -133,9 +133,10 @@ export function echoToStore(data: JSONValue): void {
   let entity = new Entity();
   entity.set("id", Value.fromString(id));
   entity.set("value", Value.fromString(value));
+  entity.set("extra", userData);
   store.set("Thing", id, entity);
 }
 
-export function ipfsMap(hash: string): void {
-  ipfs.map(hash, "echoToStore", ["json"])
+export function ipfsMap(hash: string, userData: string): void {
+  ipfs.map(hash, "echoToStore", Value.fromString(userData), ["json"])
 }