added buffered reading to tokenizer

Author: daggaz

src/json_stream/loader.py

@@ -3,9 +3,9 @@
 from json_stream.select_tokenizer import default_tokenizer
 
 
-def load(fp_or_iterable, persistent=False, tokenizer=default_tokenizer):
+def load(fp_or_iterable, persistent=False, tokenizer=default_tokenizer, buffering=-1):
     fp = ensure_file(fp_or_iterable)
-    token_stream = tokenizer(fp)
+    token_stream = tokenizer(fp, buffering=buffering)
     token_type, token = next(token_stream)
     if token_type == TokenType.OPERATOR:
         return StreamingJSONBase.factory(token, token_stream, persistent)

src/json_stream/tests/test_buffering.py

@@ -12,9 +12,9 @@ def test_buffering(self):
         self._test_buffering(tokenizer=rust_tokenizer_or_raise())
 
     def test_buffering_python_tokenizer(self):
-        self._test_buffering(tokenizer=tokenize)
+        self._test_buffering(tokenizer=tokenize, buffering=0)
 
-    def _test_buffering(self, tokenizer):
+    def _test_buffering(self, tokenizer, **load_args):
         happenings = []
 
         def data_in_chunks(data, chunk_size=15):
@@ -24,7 +24,7 @@ def data_in_chunks(data, chunk_size=15):
                 yield part
 
         json_string = b'{"tasks":[{"id":1,"title":"task1"},{"id":2,"title":"task2"},{"id":3,"title":"task3"}]}'
-        stream = json_stream.load(data_in_chunks(json_string), tokenizer=tokenizer)
+        stream = json_stream.load(data_in_chunks(json_string), tokenizer=tokenizer, **load_args)
 
         for task in stream["tasks"]:
             happenings.append(('item', to_standard_types(task)))

src/json_stream/tokenizer.py

@@ -78,7 +78,7 @@ def _ensure_text(stream):
     return stream
 
 
-def tokenize(stream):
+def tokenize(stream, *, buffering=-1, **_):
     stream = _ensure_text(stream)
 
     def is_delimiter(char):
@@ -365,9 +365,19 @@ def process_char(char):
 
         return advance, next_state
     state = State.WHITESPACE
-    c = stream.read(1)
-    index = 0
-    while c:
+    if not buffering:
+        buffering = 1
+    elif buffering <= 0:
+        buffering = io.DEFAULT_BUFFER_SIZE
+    buffering = buffering.__index__()
+    buffer = stream.read(buffering)
+    c = None
+    index = -1
+    advance = True
+    while buffer:
+        if advance:
+            c, buffer = buffer[0], buffer[1:] or stream.read(buffering)
+            index += 1
         try:
             advance, state = process_char(c)
         except ValueError as e:
@@ -376,9 +386,6 @@ def process_char(char):
             completed = False
             token = []
             yield now_token
-        if advance:
-            c = stream.read(1)
-            index += 1
     process_char(SpecialChar.EOF)
     if completed:
         yield now_token

src/json_stream/visitor.py

@@ -19,9 +19,9 @@ def _visit(obj, visitor, path):
         visitor(obj, path)
 
 
-def visit(fp_or_iterator, visitor, tokenizer=default_tokenizer):
+def visit(fp_or_iterator, visitor, tokenizer=default_tokenizer, buffering=-1):
     fp = ensure_file(fp_or_iterator)
-    token_stream = tokenizer(fp)
+    token_stream = tokenizer(fp, buffering=buffering)
     _, token = next(token_stream)
     obj = StreamingJSONBase.factory(token, token_stream, persistent=False)
     _visit(obj, visitor, ())