Add channel to pub/sub

deno2/deno.cc

@@ -133,16 +133,22 @@ void Pub(const v8::FunctionCallbackInfo<v8::Value>& args) {
   v8::Locker locker(d->isolate);
   v8::EscapableHandleScope handle_scope(isolate);
 
-  v8::Local<v8::Value> v = args[0];
+  assert(args.Length() == 2);
-  assert(v->IsArrayBuffer());
+  v8::Local<v8::Value> channel_v = args[0];
+  assert(channel_v->IsString());
+  v8::String::Utf8Value channel_vstr(isolate, channel_v);
+  const char* channel = *channel_vstr;
 
-  auto ab = v8::Local<v8::ArrayBuffer>::Cast(v);
+  v8::Local<v8::Value> ab_v = args[1];
+  assert(ab_v->IsArrayBuffer());
+
+  auto ab = v8::Local<v8::ArrayBuffer>::Cast(ab_v);
   auto contents = ab->GetContents();
 
   void* buf = contents.Data();
   int buflen = static_cast<int>(contents.ByteLength());
 
-  auto retbuf = d->cb(d, deno_buf{buf, buflen});
+  auto retbuf = d->cb(d, channel, deno_buf{buf, buflen});
   if (retbuf.data) {
     // TODO(ry) Support zero-copy.
     auto ab = v8::ArrayBuffer::New(d->isolate, retbuf.len);
@@ -270,10 +276,7 @@ bool deno_execute(Deno* d, const char* js_filename, const char* js_source) {
   return deno::Execute(context, js_filename, js_source);
 }
 
-// Routes message to the javascript callback set with deno_sub().
+bool deno_pub(Deno* d, const char* channel, deno_buf buf) {
-// False return value indicates error. Check deno_last_exception() for exception
-// text. Caller owns buf.
-bool deno_pub(Deno* d, deno_buf buf) {
   v8::Locker locker(d->isolate);
   v8::Isolate::Scope isolate_scope(d->isolate);
   v8::HandleScope handle_scope(d->isolate);
@@ -293,10 +296,9 @@ bool deno_pub(Deno* d, deno_buf buf) {
   auto ab = v8::ArrayBuffer::New(d->isolate, buf.len);
   memcpy(ab->GetContents().Data(), buf.data, buf.len);
 
-  v8::Local<v8::Value> args[1];
+  v8::Local<v8::Value> args[2];
-  args[0] = ab;
+  args[0] = deno::v8_str(channel);
-  assert(!args[0].IsEmpty());
+  args[1] = ab;
-  assert(!try_catch.HasCaught());
 
   sub->Call(context->Global(), 1, args);
 

deno2/include/deno.h

@@ -18,7 +18,7 @@ struct deno_s;
 typedef struct deno_s Deno;
 
 // The callback from V8 when data is sent.
-typedef deno_buf (*deno_sub_cb)(Deno* d, deno_buf buf);
+typedef deno_buf (*deno_sub_cb)(Deno* d, const char* channel, deno_buf buf);
 
 void deno_init();
 const char* deno_v8_version();
@@ -31,9 +31,9 @@ Deno* deno_new(void* data, deno_sub_cb cb);
 // Get error text with deno_last_exception().
 bool deno_execute(Deno* d, const char* js_filename, const char* js_source);
 
-// Returns false on error.
+// Routes message to the javascript callback set with deno_sub(). A false return
-// Get error text with deno_last_exception().
+// value indicates error. Check deno_last_exception() for exception text.
-bool deno_pub(Deno* d, deno_buf buf);
+bool deno_pub(Deno* d, const char* channel, deno_buf buf);
 
 const char* deno_last_exception(Deno* d);
 

deno2/js/mock_runtime.js

@@ -16,13 +16,15 @@ function CanCallFunction() {
 }
 
 function PubSuccess() {
-  deno_sub(msg => {
+  deno_sub((channel, msg) => {
+    assert(channel === "PubSuccess");
     deno_print("PubSuccess: ok");
   });
 }
 
 function PubByteLength() {
-  deno_sub(msg => {
+  deno_sub((channel, msg) => {
+    assert(channel === "PubByteLength");
     assert(msg instanceof ArrayBuffer);
     assert(msg.byteLength === 3);
   });
@@ -31,16 +33,16 @@ function PubByteLength() {
 function SubReturnEmpty() {
   const ui8 = new Uint8Array("abc".split("").map(c => c.charCodeAt(0)));
   const ab = typedArrayToArrayBuffer(ui8);
-  let r = deno_pub(ab);
+  let r = deno_pub("SubReturnEmpty", ab);
   assert(r == null);
-  r = deno_pub(ab);
+  r = deno_pub("SubReturnEmpty", ab);
   assert(r == null);
 }
 
 function SubReturnBar() {
   const ui8 = new Uint8Array("abc".split("").map(c => c.charCodeAt(0)));
   const ab = typedArrayToArrayBuffer(ui8);
-  const r = deno_pub(ab);
+  const r = deno_pub("SubReturnBar", ab);
   assert(r instanceof ArrayBuffer);
   assert(r.byteLength === 3);
   const rui8 = new Uint8Array(r);

deno2/mock_runtime_test.cc

@@ -31,7 +31,7 @@ deno_buf strbuf(const char* str) {
 TEST(MockRuntimeTest, PubSuccess) {
   Deno* d = deno_new(NULL, NULL);
   EXPECT_TRUE(deno_execute(d, "a.js", "PubSuccess()"));
-  EXPECT_TRUE(deno_pub(d, strbuf("abc")));
+  EXPECT_TRUE(deno_pub(d, "PubSuccess", strbuf("abc")));
   deno_dispose(d);
 }
 
@@ -39,21 +39,22 @@ TEST(MockRuntimeTest, PubByteLength) {
   Deno* d = deno_new(NULL, NULL);
   EXPECT_TRUE(deno_execute(d, "a.js", "PubByteLength()"));
   // We pub the wrong sized message, it should throw.
-  EXPECT_FALSE(deno_pub(d, strbuf("abcd")));
+  EXPECT_FALSE(deno_pub(d, "PubByteLength", strbuf("abcd")));
   deno_dispose(d);
 }
 
 TEST(MockRuntimeTest, PubNoCallback) {
   Deno* d = deno_new(NULL, NULL);
   // We didn't call deno_sub(), pubing should fail.
-  EXPECT_FALSE(deno_pub(d, strbuf("abc")));
+  EXPECT_FALSE(deno_pub(d, "PubNoCallback", strbuf("abc")));
   deno_dispose(d);
 }
 
 TEST(MockRuntimeTest, SubReturnEmpty) {
   static int count = 0;
-  Deno* d = deno_new(NULL, [](Deno* _, deno_buf buf) {
+  Deno* d = deno_new(NULL, [](auto _, auto channel, auto buf) {
     count++;
+    EXPECT_STREQ(channel, "SubReturnEmpty");
     EXPECT_EQ(static_cast<size_t>(3), buf.len);
     // TODO(ry) buf.data should just be a char*.
     char* data = reinterpret_cast<char*>(buf.data);
@@ -69,8 +70,9 @@ TEST(MockRuntimeTest, SubReturnEmpty) {
 
 TEST(MockRuntimeTest, SubReturnBar) {
   static int count = 0;
-  Deno* d = deno_new(NULL, [](Deno* _, deno_buf buf) {
+  Deno* d = deno_new(NULL, [](auto _, auto channel, auto buf) {
     count++;
+    EXPECT_STREQ(channel, "SubReturnBar");
     EXPECT_EQ(static_cast<size_t>(3), buf.len);
     // TODO(ry) buf.data should just be a char*.
     char* data = reinterpret_cast<char*>(buf.data);