This is an automated email from the ASF dual-hosted git repository.
shiro pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 72ce489 ARROW-4980: [GLib] Use GInputStream as the parent of
GArrowInputStream
72ce489 is described below
commit 72ce4892b0f1ccbc5b7adfa530ca36af5264a339
Author: Kouhei Sutou <k...@clear-code.com>
AuthorDate: Sat Mar 23 07:50:25 2019 +0900
ARROW-4980: [GLib] Use GInputStream as the parent of GArrowInputStream
If we use GInputStream as the parent of GArrowInputStream, we can use
GInputStream related feature such as encoding conversion to Arrow's
input stream.
Author: Kouhei Sutou <k...@clear-code.com>
Closes #3997 from kou/glib-gio-input-stream and squashes the following
commits:
57381937 <Kouhei Sutou> Use GInputStream as the parent of GArrowInputStream
---
c_glib/arrow-glib/input-stream.cpp | 66 ++++++++++++++++++++++++++++++---
c_glib/arrow-glib/input-stream.h | 4 +-
c_glib/test/test-buffer-input-stream.rb | 15 ++++++++
3 files changed, 77 insertions(+), 8 deletions(-)
diff --git a/c_glib/arrow-glib/input-stream.cpp
b/c_glib/arrow-glib/input-stream.cpp
index ab24186..459745d 100644
--- a/c_glib/arrow-glib/input-stream.cpp
+++ b/c_glib/arrow-glib/input-stream.cpp
@@ -65,8 +65,7 @@ typedef struct GArrowInputStreamPrivate_ {
} GArrowInputStreamPrivate;
enum {
- PROP_0,
- PROP_INPUT_STREAM
+ PROP_INPUT_STREAM = 1
};
static std::shared_ptr<arrow::io::FileInterface>
@@ -100,7 +99,7 @@
garrow_input_stream_readable_interface_init(GArrowReadableInterface *iface)
G_DEFINE_TYPE_WITH_CODE(GArrowInputStream,
garrow_input_stream,
- G_TYPE_OBJECT,
+ G_TYPE_INPUT_STREAM,
G_ADD_PRIVATE(GArrowInputStream)
G_IMPLEMENT_INTERFACE(GARROW_TYPE_FILE,
garrow_input_stream_file_interface_init)
@@ -154,6 +153,58 @@ garrow_input_stream_get_property(GObject *object,
}
}
+static gssize
+garrow_input_stream_read(GInputStream *stream,
+ void *buffer,
+ gsize count,
+ GCancellable *cancellable,
+ GError **error)
+{
+ if (g_cancellable_set_error_if_cancelled(cancellable, error)) {
+ return -1;
+ }
+ auto arrow_input_stream =
+ garrow_input_stream_get_raw(GARROW_INPUT_STREAM(stream));
+ int64_t n_read_bytes;
+ auto status = arrow_input_stream->Read(count, &n_read_bytes, buffer);
+ if (!garrow_error_check(error, status, "[input-stream][read]")) {
+ return -1;
+ }
+ return n_read_bytes;
+}
+
+static gssize
+garrow_input_stream_skip(GInputStream *stream,
+ gsize count,
+ GCancellable *cancellable,
+ GError **error)
+{
+ if (g_cancellable_set_error_if_cancelled(cancellable, error)) {
+ return -1;
+ }
+ auto arrow_input_stream =
+ garrow_input_stream_get_raw(GARROW_INPUT_STREAM(stream));
+ auto status = arrow_input_stream->Advance(count);
+ if (!garrow_error_check(error, status, "[input-stream][skip]")) {
+ return -1;
+ }
+ return count;
+}
+
+static gboolean
+garrow_input_stream_close(GInputStream *stream,
+ GCancellable *cancellable,
+ GError **error)
+{
+ if (g_cancellable_set_error_if_cancelled(cancellable, error)) {
+ return FALSE;
+ }
+ auto arrow_input_stream =
+ garrow_input_stream_get_raw(GARROW_INPUT_STREAM(stream));
+ auto status = arrow_input_stream->Close();
+ return garrow_error_check(error, status, "[input-stream][close]");
+}
+
static void
garrow_input_stream_init(GArrowInputStream *object)
{
@@ -163,11 +214,15 @@ static void
garrow_input_stream_class_init(GArrowInputStreamClass *klass)
{
auto gobject_class = G_OBJECT_CLASS(klass);
-
gobject_class->finalize = garrow_input_stream_finalize;
gobject_class->set_property = garrow_input_stream_set_property;
gobject_class->get_property = garrow_input_stream_get_property;
+ auto input_stream_class = G_INPUT_STREAM_CLASS(klass);
+ input_stream_class->read_fn = garrow_input_stream_read;
+ input_stream_class->skip = garrow_input_stream_skip;
+ input_stream_class->close_fn = garrow_input_stream_close;
+
GParamSpec *spec;
spec = g_param_spec_pointer("input-stream",
"Input stream",
@@ -356,8 +411,7 @@ typedef struct GArrowBufferInputStreamPrivate_ {
} GArrowBufferInputStreamPrivate;
enum {
- PROP_0_,
- PROP_BUFFER
+ PROP_BUFFER = 1,
};
G_DEFINE_TYPE_WITH_PRIVATE(GArrowBufferInputStream,
diff --git a/c_glib/arrow-glib/input-stream.h b/c_glib/arrow-glib/input-stream.h
index 745b912..8f86741 100644
--- a/c_glib/arrow-glib/input-stream.h
+++ b/c_glib/arrow-glib/input-stream.h
@@ -32,10 +32,10 @@ G_DECLARE_DERIVABLE_TYPE(GArrowInputStream,
garrow_input_stream,
GARROW,
INPUT_STREAM,
- GObject)
+ GInputStream)
struct _GArrowInputStreamClass
{
- GObjectClass parent_class;
+ GInputStreamClass parent_class;
};
gboolean garrow_input_stream_advance(GArrowInputStream *input_stream,
diff --git a/c_glib/test/test-buffer-input-stream.rb
b/c_glib/test/test-buffer-input-stream.rb
index cb6a667..5a9bc4d 100644
--- a/c_glib/test/test-buffer-input-stream.rb
+++ b/c_glib/test/test-buffer-input-stream.rb
@@ -47,4 +47,19 @@ class TestBufferInputStream < Test::Unit::TestCase
assert_equal(buffer_input_stream.read(5).data.to_s,
peeked_data.to_s)
end
+
+ def test_gio_input_stream
+ # U+3042 HIRAGANA LETTER A
+ data = "\u3042"
+ convert_encoding = "cp932"
+ buffer = Arrow::Buffer.new(data)
+ buffer_input_stream = Arrow::BufferInputStream.new(buffer)
+ converter = Gio::CharsetConverter.new(convert_encoding, "UTF-8")
+ convert_input_stream =
+ Gio::ConverterInputStream.new(buffer_input_stream, converter)
+ gio_input_stream = Arrow::GIOInputStream.new(convert_input_stream)
+ raw_read_data = gio_input_stream.read(10).data.to_s
+ assert_equal(data.encode(convert_encoding),
+ raw_read_data.dup.force_encoding(convert_encoding))
+ end
end
