RPC: fix merge conflicts

libraries/RPC/examples/PortentaX8_EchoServer/PortentaX8_EchoServer.ino

@@ -1,21 +1,24 @@
-#include "RPC.h"
 #include "SerialRPC.h"
+#include "RPC.h"
 
 /*
- * This sketch demonstrates how to interact with the Portenta X8 Serial port (over USB)
- * On the board, launch both 'proxy' and 'example' binaries (from https://github.com/arduino/portentax8-m4-proxy)
- * The M4 provides the 'subtract' API (which will be invoked by 'example'
- * It also provides a full duplex Serial-like interface that is proxies through the serial monitor
- * Last but not leas, when you write 'echo' the corresponding function in 'example' will be triggered
- */
+   This sketch demonstrates how to interact with the Portenta X8 Serial port (over USB)
+   On the board, launch both 'proxy' and 'example' binaries (from https://github.com/bcmi-labs/portentax8-m4-proxy)
+   The M4 provides the 'subtract' API (which will be invoked by 'example'
+   It also provides a full duplex Serial-like interface that is proxies through the serial monitor
+   Last but not leas, when you write 'echo' the corresponding function in 'example' will be triggered
+*/
 
 int subtract(int a, int b) {
-  return a-b;
+  return a - b;
 }
 
+int led_status = 0;
+
 void setup() {
   // put your setup code here, to run once:
   Serial.begin(115200);
+  pinMode(LED_BUILTIN, OUTPUT);
   RPC.bind("subtract", subtract);
   delay(1000);
 }
@@ -26,17 +29,44 @@ void loop() {
   //RPC.print("hello");
   //RPC.send("echo", "test");
   //auto res = RPC.call("add", 5, 8).as<int>();
-  //RPC.send("echo", String(res).c_str());
+  if (millis() % 1000 == 0) {
+    Serial.println("loop");
+    delay(2);
+  }
 
   String str = "";
   while (Serial.available()) {
     str += (char)Serial.read();
   }
   if (str != "") {
-    Serial.print(str);
+    //Serial.print(str);
+  }
+
+  if (str.startsWith("whoami")) {
+    digitalWrite(LED_BUILTIN, HIGH);
+    auto res = RPC.call("whoami").as<std::string>();
+    Serial.println(res.c_str());
+    digitalWrite(LED_BUILTIN, LOW);
   }
+
+  if (str.startsWith("divide")) {
+    float a = random() % 15000;
+    float b = random() % 15000;
+    Serial.println(String(a) + " / " + String(b));
+    auto res = RPC.call("divide", a, b).as<float>();
+    Serial.println(String(a) + " / " + String(b) + " = " + String(res));
+  }
+
+  if (str.startsWith("add")) {
+    int a = random() % 15000;
+    int b = random() % 15000;
+    Serial.println(String(a) + " + " + String(b));
+    auto res = RPC.call("add", a, b).as<int>();
+    Serial.println(String(a) + " + " + String(b) + " = " + String(res));
+  }
+
   if (str.startsWith("echo")) {
-    delay(100);
-    RPC.send("echo", "test");
+    auto res = RPC.call("echo", "X8").as<std::string>();
+    Serial.println(res.c_str());
   }
-}
+}

libraries/RPC/src/RPC.cpp

@@ -12,6 +12,9 @@ arduino::RPCClass RPC;
 osThreadId eventHandlerThreadId;
 static rtos::Mutex mutex;
 static struct rpmsg_endpoint endpoints[2];
+#ifdef CORE_CM4
+static bool endpoints_init[2] = { 0 };
+#endif
 
 void RPCClass::new_service_cb(struct rpmsg_device *rdev, const char *name, uint32_t dest) {
     uint8_t buffer[1] = {0};
@@ -31,20 +34,23 @@ void RPCClass::new_service_cb(struct rpmsg_device *rdev, const char *name, uint3
 
 int RPCClass::rpmsg_recv_callback(struct rpmsg_endpoint *ept, void *data, size_t len, uint32_t src, void *priv) {
     #ifdef CORE_CM4
-    static bool ep_init_msg[2] = { 0 };
-    if (!ep_init_msg[ENDPOINT_ID_RPC] && ept == &endpoints[ENDPOINT_ID_RPC]) {
-        ep_init_msg[ENDPOINT_ID_RPC] = true;
+    if (!endpoints_init[ENDPOINT_ID_RPC] && ept == &endpoints[ENDPOINT_ID_RPC]) {
+        endpoints_init[ENDPOINT_ID_RPC] = true;
         return 0;
-    } else if (!ep_init_msg[ENDPOINT_ID_RAW] && ept == &endpoints[ENDPOINT_ID_RAW]) {
-        ep_init_msg[ENDPOINT_ID_RAW] = true;
+    } else if (!endpoints_init[ENDPOINT_ID_RAW] && ept == &endpoints[ENDPOINT_ID_RAW]) {
+        endpoints_init[ENDPOINT_ID_RAW] = true;
         return 0;
     }
     #endif
 
     if (ept == &endpoints[ENDPOINT_ID_RAW]) {
         // data on raw endpoint
         if (RPC.raw_callback) {
-            RPC.raw_callback.call((const uint8_t *) data, len);
+            RPC.raw_callback.call((uint8_t *) data, len);
+        } else {
+            for (size_t i=0; i<len; i++) {
+                RPC.rx_buffer.store_char(((uint8_t *) data)[i]);
+            }
         }
         return 0;
     }
@@ -176,6 +182,16 @@ int RPCClass::begin() {
         return 0;
     }
 
+    // Wait for endpoints to be initialized first by the host before allowing
+    // the remote to use the endpoints.
+    uint32_t millis_start = millis();
+    while (!endpoints_init[ENDPOINT_ID_RPC] || !endpoints_init[ENDPOINT_ID_RAW]) {
+        if ((millis() - millis_start) >= 5000) {
+            return 0;
+        }
+        osDelay(10);
+    }
+
     return 1;
 }
 #endif
@@ -217,53 +233,25 @@ void RPCClass::request(uint8_t *buf, size_t len) {
     RPCLIB_MSGPACK::unpacked result;
     while (unpacker.next(result)) {
         auto msg = result.get();
-        if (msg.via.array.size == 1) {
-            // raw array
-            std::tuple<RPCLIB_MSGPACK::object> arr;
-            msg.convert(arr);
-
-            std::vector<uint8_t> buf;
-            std::get<0>(arr).convert(buf);
-
-            for (size_t i = 0; i < buf.size(); i++) {
-                rx_buffer.store_char(buf[i]);
-            }
-        }
-
-        if (msg.via.array.size > 2) {
-            auto resp = rpc::detail::dispatcher::dispatch(msg, false);
-            auto data = resp.get_data();
-            if (!resp.is_empty()) {
-                OPENAMP_send(&endpoints[ENDPOINT_ID_RPC], data.data(), data.size());
-            }
+        auto resp = rpc::detail::dispatcher::dispatch(msg, false);
+        auto data = resp.get_data();
+        if (!resp.is_empty()) {
+            OPENAMP_send(&endpoints[ENDPOINT_ID_RPC], data.data(), data.size());
         }
     }
 }
 
 size_t RPCClass::write(uint8_t c) {
-    return write(&c, 1, true, false);
+    return write(&c, 1, true);
 }
 
 void rpc::client::write(RPCLIB_MSGPACK::sbuffer *buffer) {
-    RPC.write((const uint8_t *) buffer->data(), buffer->size(), false, false);
+    RPC.write((const uint8_t *) buffer->data(), buffer->size(), false);
 }
 
-size_t RPCClass::write(const uint8_t *buf, size_t len, bool pack, bool raw) {
-    if (pack) {
-        std::vector<uint8_t> tx_buffer;
-        for (size_t i = 0; i < len; i++) {
-            tx_buffer.push_back(buf[i]);
-        }
-        auto call_obj = std::make_tuple(tx_buffer);
-
-        RPCLIB_MSGPACK::sbuffer buffer;
-        RPCLIB_MSGPACK::pack(buffer, call_obj);
-        len = buffer.size();
-        buf = (const uint8_t *) buffer.data();
-    }
-
+size_t RPCClass::write(const uint8_t *buf, size_t len, bool raw) {
     mutex.lock();
     OPENAMP_send(&endpoints[raw ? ENDPOINT_ID_RAW : ENDPOINT_ID_RPC], buf, len);
     mutex.unlock();
     return len;
-}
+}

libraries/RPC/src/RPC.h

@@ -56,7 +56,7 @@ class RPCClass : public Stream, public rpc::detail::dispatcher {
         }
         void flush(void) {};
         size_t write(uint8_t c);
-        size_t write(const uint8_t *buf, size_t len, bool pack = true, bool raw = false);
+        size_t write(const uint8_t *buf, size_t len, bool raw = true);
 
         using Print::write; // pull in write(str) and write(buf, size) from Print
         operator bool() {
@@ -115,14 +115,14 @@ class RPCClass : public Stream, public rpc::detail::dispatcher {
         }
 
         rpc::client* clients[10];
+        RingBufferN<512> rx_buffer;
         mbed::Callback<void(const uint8_t *buf, size_t len)> raw_callback;
 
     private:
         bool initialized = false;
         uint32_t _timeout = osWaitForever;
         bool has_timed_out = false;
         rtos::Thread* eventThread;
-        RingBufferN<512> rx_buffer;
         RPCLIB_MSGPACK::unpacker unpacker;
 
         static void new_service_cb(struct rpmsg_device *rdev, const char *name, uint32_t dest);