Message

Author: nealrichardson

r/NAMESPACE

@@ -7,8 +7,6 @@ S3method("==","arrow::Message")
 S3method("==","arrow::RecordBatch")
 S3method("==","arrow::Schema")
 S3method("==",Array)
-S3method(MessageReader,InputStream)
-S3method(MessageReader,default)
 S3method(RecordBatchFileReader,Buffer)
 S3method(RecordBatchFileReader,RandomAccessFile)
 S3method(RecordBatchFileReader,character)
@@ -31,12 +29,12 @@ S3method(parquet_file_reader,RandomAccessFile)
 S3method(parquet_file_reader,character)
 S3method(parquet_file_reader,raw)
 S3method(print,"arrow-enum")
-S3method(read_message,"arrow::MessageReader")
 S3method(read_message,InputStream)
+S3method(read_message,MessageReader)
 S3method(read_message,default)
-S3method(read_record_batch,"arrow::Message")
 S3method(read_record_batch,Buffer)
 S3method(read_record_batch,InputStream)
+S3method(read_record_batch,Message)
 S3method(read_record_batch,raw)
 S3method(read_schema,"arrow::Message")
 S3method(read_schema,Buffer)

r/R/enums.R

@@ -69,7 +69,7 @@ FileMode <- enum("FileMode",
 
 #' @rdname enums
 #' @export
-MessageType <- enum("arrow::Message::Type",
+MessageType <- enum("Message::Type",
   NONE = 0L, SCHEMA = 1L, DICTIONARY_BATCH = 2L, RECORD_BATCH = 3L, TENSOR = 4L
 )
 

r/R/message.R

@@ -29,10 +29,10 @@
 #'
 #' @rdname arrow__ipc__Message
 #' @name arrow__ipc__Message
-`arrow::Message` <- R6Class("arrow::Message", inherit = Object,
+Message <- R6Class("Message", inherit = Object,
   public = list(
     Equals = function(other){
-      assert_that(inherits(other, "arrow::Message"))
+      assert_that(inherits(other, "Message"))
       ipc___Message__Equals(self, other)
     },
     body_length = function() ipc___Message__body_length(self),
@@ -60,29 +60,18 @@
 #'
 #' @rdname arrow__ipc__MessageReader
 #' @name arrow__ipc__MessageReader
-`arrow::MessageReader` <- R6Class("arrow::MessageReader", inherit = Object,
+#' @export
+MessageReader <- R6Class("MessageReader", inherit = Object,
   public = list(
-    ReadNextMessage = function() unique_ptr(`arrow::Message`, ipc___MessageReader__ReadNextMessage(self))
+    ReadNextMessage = function() unique_ptr(Message, ipc___MessageReader__ReadNextMessage(self))
   )
 )
 
-#' Open a MessageReader that reads from a stream
-#'
-#' @param stream an InputStream
-#'
-#' @export
-MessageReader <- function(stream) {
-  UseMethod("MessageReader")
-}
-
-#' @export
-MessageReader.default <- function(stream) {
-  MessageReader(BufferReader$create(stream))
-}
-
-#' @export
-MessageReader.InputStream <- function(stream) {
-  unique_ptr(`arrow::MessageReader`, ipc___MessageReader__Open(stream))
+MessageReader$create <- function(stream) {
+  if (!inherits(stream, "InputStream")) {
+    stream <- BufferReader$create(stream)
+  }
+  unique_ptr(MessageReader, ipc___MessageReader__Open(stream))
 }
 
 #' Read a Message from a stream
@@ -95,16 +84,16 @@ read_message <- function(stream) {
 }
 
 #' @export
-read_message.default<- function(stream) {
+read_message.default <- function(stream) {
   read_message(BufferReader$create(stream))
 }
 
 #' @export
 read_message.InputStream <- function(stream) {
-  unique_ptr(`arrow::Message`, ipc___ReadMessage(stream) )
+  unique_ptr(Message, ipc___ReadMessage(stream) )
 }
 
 #' @export
-`read_message.arrow::MessageReader` <- function(stream) {
+read_message.MessageReader <- function(stream) {
   stream$ReadNextMessage()
 }

r/R/read_record_batch.R

@@ -28,7 +28,7 @@ read_record_batch <- function(obj, schema){
 }
 
 #' @export
-`read_record_batch.arrow::Message` <- function(obj, schema) {
+read_record_batch.Message <- function(obj, schema) {
   assert_that(inherits(schema, "arrow::Schema"))
   shared_ptr(`arrow::RecordBatch`, ipc___ReadRecordBatch__Message__Schema(obj, schema))
 }
@@ -47,7 +47,7 @@ read_record_batch.raw <- function(obj, schema){
 }
 
 #' @export
-`read_record_batch.Buffer` <- function(obj, schema){
+read_record_batch.Buffer <- function(obj, schema){
   stream <- BufferReader$create(obj)
   on.exit(stream$close())
   read_record_batch(stream, schema)

r/man/MessageReader.Rd

@@ -15,15 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 
-context("arrow::Message")
+context("Message")
 
 test_that("read_message can read from input stream", {
   batch <- record_batch(x = 1:10)
   bytes <- batch$serialize()
   stream <- BufferReader$create(bytes)
 
   message <- read_message(stream)
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$RECORD_BATCH)
   expect_is(message$body, "Buffer")
   expect_is(message$metadata, "Buffer")
@@ -37,7 +37,7 @@ test_that("read_message() can read Schema messages", {
   stream <- BufferReader$create(bytes)
   message <- read_message(stream)
 
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$SCHEMA)
   expect_is(message$body, "Buffer")
   expect_is(message$metadata, "Buffer")

r/man/arrow__ipc__Message.Rd

@@ -15,16 +15,16 @@
 # specific language governing permissions and limitations
 # under the License.
 
-context("arrow::MessageReader")
+context("MessageReader")
 
 test_that("MessageReader can be created from raw vectors", {
   batch <- record_batch(x = 1:10)
   bytes <- batch$serialize()
 
-  reader <- MessageReader(bytes)
+  reader <- MessageReader$create(bytes)
 
   message <- reader$ReadNextMessage()
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$RECORD_BATCH)
   expect_is(message$body, "Buffer")
   expect_is(message$metadata, "Buffer")
@@ -35,10 +35,10 @@ test_that("MessageReader can be created from raw vectors", {
   schema <- schema(x = int32())
   bytes <- schema$serialize()
 
-  reader <- MessageReader(bytes)
+  reader <- MessageReader$create(bytes)
 
   message <- reader$ReadNextMessage()
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$SCHEMA)
   expect_is(message$body, "Buffer")
   expect_is(message$metadata, "Buffer")
@@ -54,11 +54,11 @@ test_that("MessageReader can be created from input stream", {
   stream <- BufferReader$create(bytes)
   expect_is(stream, "BufferReader")
 
-  reader <- MessageReader(stream)
-  expect_is(reader, "arrow::MessageReader")
+  reader <- MessageReader$create(stream)
+  expect_is(reader, "MessageReader")
 
   message <- reader$ReadNextMessage()
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$RECORD_BATCH)
   expect_is(message$body, "Buffer")
   expect_is(message$metadata, "Buffer")
@@ -72,11 +72,11 @@ test_that("MessageReader can be created from input stream", {
   stream <- BufferReader$create(bytes)
   expect_is(stream, "BufferReader")
 
-  reader <- MessageReader(stream)
-  expect_is(reader, "arrow::MessageReader")
+  reader <- MessageReader$create(stream)
+  expect_is(reader, "MessageReader")
 
   message <- reader$ReadNextMessage()
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$SCHEMA)
   expect_is(message$body, "Buffer")
   expect_is(message$metadata, "Buffer")

r/man/arrow__ipc__MessageReader.Rd

@@ -38,17 +38,17 @@ test_that("reading schema from Buffer", {
   buffer <- stream$getvalue()
   expect_is(buffer, "Buffer")
 
-  reader <- MessageReader(buffer)
-  expect_is(reader, "arrow::MessageReader")
+  reader <- MessageReader$create(buffer)
+  expect_is(reader, "MessageReader")
 
   message <- reader$ReadNextMessage()
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$SCHEMA)
 
   stream <- BufferReader$create(buffer)
   expect_is(stream, "BufferReader")
   message <- read_message(stream)
-  expect_is(message, "arrow::Message")
+  expect_is(message, "Message")
   expect_equal(message$type, MessageType$SCHEMA)
 })