You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by da...@apache.org on 2017/01/30 18:00:31 UTC
[4/7] tinkerpop git commit: added example driver code to branch
corresponding to JIRA issue
added example driver code to branch corresponding to JIRA issue
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/2577a13d
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/2577a13d
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/2577a13d
Branch: refs/heads/TINKERPOP-1599
Commit: 2577a13d6a9fe68ef01975736c33b1f7b7c6a84c
Parents: a98fb13
Author: davebshow <da...@gmail.com>
Authored: Tue Jan 24 16:52:53 2017 -0500
Committer: davebshow <da...@gmail.com>
Committed: Mon Jan 30 11:51:24 2017 -0500
----------------------------------------------------------------------
.../main/jython/gremlin_python/driver/client.py | 112 ++++++++
.../jython/gremlin_python/driver/connection.py | 75 ++++++
.../driver/driver_remote_connection.py | 258 +++----------------
.../jython/gremlin_python/driver/protocol.py | 98 +++++++
.../gremlin_python/driver/remote_connection.py | 95 ++++---
.../jython/gremlin_python/driver/request.py | 23 ++
.../jython/gremlin_python/driver/resultset.py | 75 ++++++
.../jython/gremlin_python/driver/serializer.py | 115 +++++++++
.../gremlin_python/driver/tornado/__init__.py | 18 ++
.../gremlin_python/driver/tornado/transport.py | 46 ++++
.../jython/gremlin_python/driver/transport.py | 44 ++++
.../jython/gremlin_python/process/traversal.py | 16 +-
gremlin-python/src/main/jython/setup.py | 19 +-
13 files changed, 727 insertions(+), 267 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/client.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/client.py b/gremlin-python/src/main/jython/gremlin_python/driver/client.py
new file mode 100644
index 0000000..0c1ba6c
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/client.py
@@ -0,0 +1,112 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+import collections
+import functools
+
+from concurrent.futures import ThreadPoolExecutor
+
+from six.moves import queue
+
+from gremlin_python.driver import connection, protocol, request
+from gremlin_python.process import traversal
+
+# This is until concurrent.futures backport 3.1.0 release
+try:
+ from multiprocessing import cpu_count
+except ImportError:
+ # some platforms don't have multiprocessing
+ def cpu_count():
+ return None
+
+
+class Client:
+
+ def __init__(self, url, traversal_source, protocol_factory=None,
+ transport_factory=None, pool_size=None, max_workers=None,
+ message_serializer=None, username="", password=""):
+ self._url = url
+ self._traversal_source = traversal_source
+ self._message_serializer = message_serializer
+ self._username = username
+ self._password = password
+ if transport_factory is None:
+ try:
+ from gremlin_python.driver.tornado.transport import (
+ TornadoTransport)
+ except ImportError:
+ raise Exception("Please install Tornado or pass"
+ "custom transport factory")
+ else:
+ transport_factory = lambda: TornadoTransport()
+ self._transport_factory = transport_factory
+ if protocol_factory is None:
+ protocol_factory = lambda: protocol.GremlinServerWSProtocol(
+ message_serializer=self._message_serializer,
+ username=self._username,
+ password=self._password)
+ self._protocol_factory = protocol_factory
+ if pool_size is None:
+ pool_size = 4
+ self._pool_size = pool_size
+ # This is until concurrent.futures backport 3.1.0 release
+ if max_workers is None:
+ # Use this number because ThreadPoolExecutor is often
+ # used to overlap I/O instead of CPU work.
+ max_workers = (cpu_count() or 1) * 5
+ self._executor = ThreadPoolExecutor(max_workers=max_workers)
+ # Threadsafe queue
+ self._pool = queue.Queue()
+ self._fill_pool()
+
+ @property
+ def executor(self):
+ return self._executor
+
+ @property
+ def traversal_source(self):
+ return self._traversal_source
+
+ def _fill_pool(self):
+ for i in range(self._pool_size):
+ conn = self._get_connection()
+ self._pool.put_nowait(conn)
+
+ def close(self):
+ while not self._pool.empty():
+ conn = self._pool.get(True)
+ conn.close()
+ self._executor.shutdown()
+
+ def _get_connection(self):
+ protocol = self._protocol_factory()
+ return connection.Connection(
+ self._url, self._traversal_source, protocol,
+ self._transport_factory, self._executor, self._pool)
+
+ def submit(self, message):
+ return self.submitAsync(message).result()
+
+ def submitAsync(self, message):
+ if isinstance(message, traversal.Bytecode):
+ message = request.RequestMessage(
+ processor='traversal', op='bytecode',
+ args={'gremlin': message,
+ 'aliases': {'g': self._traversal_source}})
+ conn = self._pool.get(True)
+ return conn.write(message)
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/connection.py
new file mode 100644
index 0000000..2f59883
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/connection.py
@@ -0,0 +1,75 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+import uuid
+from concurrent.futures import Future
+from six.moves import queue
+
+from gremlin_python.driver import resultset
+
+
+class Connection:
+
+ def __init__(self, url, traversal_source, protocol, transport_factory,
+ executor, pool):
+ self._url = url
+ self._traversal_source = traversal_source
+ self._protocol = protocol
+ self._transport_factory = transport_factory
+ self._executor = executor
+ self._transport = None
+ self._pool = pool
+ self._results = {}
+ self.connect()
+
+ def connect(self):
+ if self._transport:
+ self._transport.close()
+ self._transport = self._transport_factory()
+ self._transport.connect(self._url)
+ self._protocol.connection_made(self._transport)
+
+ def close(self):
+ self._transport.close()
+
+ def write(self, request_message):
+ request_id = str(uuid.uuid4())
+ result_set = resultset.ResultSet(queue.Queue(), request_id)
+ self._results[request_id] = result_set
+ # Create write task
+ future = Future()
+ future_write = self._executor.submit(
+ self._protocol.write, request_id, request_message)
+
+ def cb(f):
+ try:
+ f.result()
+ except Exception as e:
+ future.set_exception(e)
+ else:
+ # Start receive task
+ self._executor.submit(self._receive)
+ future.set_result(result_set)
+
+ future_write.add_done_callback(cb)
+ return future
+
+ def _receive(self):
+ data = self._transport.read()
+ self._protocol.data_received(data, self._results)
+ self._pool.put_nowait(self)
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
index 7bc792f..4e70a87 100644
--- a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
@@ -16,236 +16,46 @@ KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
"""
-import base64
-import functools
-import json
-import uuid
-from tornado import gen
-from tornado import concurrent
-from tornado import ioloop
-from tornado import websocket
+from concurrent.futures import Future
-from gremlin_python.structure.io.graphson import GraphSONReader, GraphSONWriter
-from .remote_connection import RemoteConnection
-from .remote_connection import RemoteTraversal
-from .remote_connection import RemoteTraversalSideEffects
+from gremlin_python.driver import client
+from gremlin_python.driver.remote_connection import (
+ RemoteTraversal, RemoteTraversalSideEffects)
-class GremlinServerError(Exception):
- pass
+class DriverRemoteConnection:
+ def __init__(self, url, traversal_source, protocol_factory=None,
+ transport_factory=None, pool_size=None, max_workers=None,
+ username="", password=""):
+ self._client = client.Client(url, traversal_source, protocol_factory,
+ transport_factory, pool_size, max_workers,
+ None, username, password)
-class DriverRemoteConnection(RemoteConnection):
- def __init__(self, url, traversal_source, username="", password="", loop=None, graphson_reader=None, graphson_writer=None):
- super(DriverRemoteConnection, self).__init__(url, traversal_source)
- self._url = url
- self._username = username
- self._password = password
- if loop is None:
- loop = ioloop.IOLoop.current()
- self._loop = loop
- self._websocket = self._loop.run_sync(lambda: websocket.websocket_connect(self.url))
- self._graphson_reader = graphson_reader or GraphSONReader()
- self._graphson_writer = graphson_writer or GraphSONWriter()
+ def close(self):
+ self._client.close()
def submit(self, bytecode):
- '''
- :param bytecode: the bytecode of a traversal to submit to the RemoteConnection
- :return: a RemoteTraversal with RemoteTraversalSideEffects
- '''
- request_id = str(uuid.uuid4())
- traversers = self._loop.run_sync(lambda: self.submit_traversal_bytecode(request_id, bytecode))
- keys, value, close = self._get_side_effect_lambdas(request_id)
- return RemoteTraversal(iter(traversers), RemoteTraversalSideEffects(keys, value, close, self._loop))
-
- def submit_async(self, bytecode):
- request_id = str(uuid.uuid4())
- future_traversers = self.submit_traversal_bytecode(request_id, bytecode)
- keys, value, close = self._get_side_effect_lambdas(request_id)
- side_effects = RemoteTraversalSideEffects(keys, value, close, self._loop)
- return RemoteTraversal(future_traversers, side_effects)
-
- @gen.coroutine
- def submit_traversal_bytecode(self, request_id, bytecode):
- message = {
- "requestId": {
- "@type": "g:UUID",
- "@value": request_id
- },
- "op": "bytecode",
- "processor": "traversal",
- "args": {
- "gremlin": self._graphson_writer.toDict(bytecode),
- "aliases": {"g": self.traversal_source}
- }
- }
- traversers = yield self._execute_message(message)
- raise gen.Return(traversers)
-
- @gen.coroutine
- def submit_sideEffect_keys(self, request_id):
- message = {
- "requestId": {
- "@type": "g:UUID",
- "@value": str(uuid.uuid4())
- },
- "op": "keys",
- "processor": "traversal",
- "args": {
- "sideEffect": {
- "@type": "g:UUID",
- "@value": request_id
- }
- }
- }
- keys = yield self._execute_message(message)
- raise gen.Return(set(keys))
-
- @gen.coroutine
- def submit_sideEffect_value(self, request_id, key):
- message = {
- "requestId": {
- "@type": "g:UUID",
- "@value": str(uuid.uuid4())
- },
- "op": "gather",
- "processor": "traversal",
- "args": {
- "sideEffect": {
- "@type": "g:UUID",
- "@value": request_id
- },
- "sideEffectKey": key,
- "aliases": {"g": self.traversal_source}
- }
- }
- try:
- value = yield self._execute_message(message)
- except:
- raise KeyError(key)
- raise gen.Return(value)
-
- @gen.coroutine
- def submit_sideEffect_close(self, request_id):
- message = {
- "requestId": {
- "@type": "g:UUID",
- "@value": str(uuid.uuid4())
- },
- "op": "close",
- "processor": "traversal",
- "args": {
- "sideEffect": {
- "@type": "g:UUID",
- "@value": request_id
- }
- }
- }
- result = yield self._execute_message(message)
- raise gen.Return(result)
-
- def _get_side_effect_lambdas(self, request_id):
- side_effect_keys = lambda: self._loop.run_sync(lambda: self.submit_sideEffect_keys(request_id))
- side_effect_value = lambda key: self._loop.run_sync(lambda: self.submit_sideEffect_value(request_id, key))
- side_effect_close = lambda: self._loop.run_sync(lambda: self.submit_sideEffect_close(request_id))
- return side_effect_keys, side_effect_value, side_effect_close
-
- @gen.coroutine
- def _execute_message(self, send_message):
- send_message = b"".join([b"\x21",
- b"application/vnd.gremlin-v3.0+json",
- json.dumps(send_message, separators=(',', ':')).encode("utf-8")])
- if self._websocket.protocol is None:
- self._websocket = yield websocket.websocket_connect(self.url)
- self._websocket.write_message(send_message, binary=True)
- response = Response(self._websocket, self._username, self._password, self._graphson_reader)
- results = None
- while True:
- recv_message = yield response.receive()
- if recv_message is None:
- break
- aggregateTo = recv_message[0]
- # on first message, get the right result data structure
- if None == results:
- if "list" == aggregateTo:
- results = []
- elif "set" == aggregateTo:
- results = set()
- elif aggregateTo in ["map", "bulkset"]:
- results = {}
- elif "none" == aggregateTo:
- results = None
- else:
- results = []
-
- # if there is no update to a structure, then the item is the result
- if results is None:
- results = recv_message[1][0]
- # updating a map is different than a list or a set
- elif isinstance(results, dict):
- if "map" == aggregateTo:
- for item in recv_message[1]:
- results.update(item)
- else:
- for item in recv_message[1]:
- results[item.object] = item.bulk
- # flat add list to result list
+ result_set = self._client.submit(bytecode)
+ results = result_set.all().result()
+ side_effects = RemoteTraversalSideEffects(result_set.request_id,
+ self._client)
+ return RemoteTraversal(iter(results), side_effects)
+
+ def submitAsync(self, bytecode):
+ future = Future()
+ future_result_set = self._client.submitAsync(bytecode)
+
+ def cb(f):
+ try:
+ result_set = f.result()
+ except Exception as e:
+ future.set_exception(e)
else:
- results += recv_message[1]
- raise gen.Return([] if None == results else results)
-
- def close(self):
- self._websocket.close()
-
-
-class Response:
- def __init__(self, websocket, username, password, graphson_reader):
- self._websocket = websocket
- self._username = username
- self._password = password
- self._closed = False
- self._graphson_reader = graphson_reader
-
- @gen.coroutine
- def receive(self):
- if self._closed:
- return
- recv_message = yield self._websocket.read_message()
- recv_message = json.loads(recv_message.decode('utf-8'))
- status_code = recv_message["status"]["code"]
- aggregateTo = recv_message["result"]["meta"].get("aggregateTo", "list")
+ results = result_set.all().result()
+ side_effects = RemoteTraversalSideEffects(result_set.request_id,
+ self._client)
+ future.set_result(RemoteTraversal(iter(results), side_effects))
- # authentification required then
- if status_code == 407:
- self._websocket.write_message(
- b"".join([b"\x21",
- b"application/vnd.gremlin-v3.0+json",
- json.dumps({
- "requestId": {
- "@type": "g:UUID",
- "@value": str(uuid.uuid4())
- },
- "op": "authentication",
- "processor": "traversal",
- "args": {
- "sasl": base64.b64encode(
- b"".join([b"\x00", self._username.encode("utf-8"),
- b"\x00", self._password.encode("utf-8")])).decode()
- }
- }, separators=(',', ':')).encode("utf-8")]), binary=True)
- results = yield self.receive()
- raise gen.Return(results)
- elif status_code == 204:
- self._closed = True
- return
- elif status_code in [200, 206]:
- results = []
- for item in recv_message["result"]["data"]:
- results.append(self._graphson_reader.toObject(item))
- if status_code == 200:
- self._closed = True
- raise gen.Return((aggregateTo, results))
- else:
- self._closed = True
- raise GremlinServerError(
- "{0}: {1}".format(status_code, recv_message["status"]["message"]))
+ future_result_set.add_done_callback(cb)
+ return future
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/protocol.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/protocol.py b/gremlin-python/src/main/jython/gremlin_python/driver/protocol.py
new file mode 100644
index 0000000..6ea17a5
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/protocol.py
@@ -0,0 +1,98 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+import abc
+import base64
+import collections
+import json
+import uuid
+
+import six
+
+from gremlin_python.driver import serializer, request
+
+
+@six.add_metaclass(abc.ABCMeta)
+class AbstractBaseProtocol:
+
+ @abc.abstractmethod
+ def connection_made(self, transport):
+ self._transport = transport
+
+ @abc.abstractmethod
+ def data_received(self, message):
+ pass
+
+ @abc.abstractmethod
+ def write(self, request_id, request_message):
+ pass
+
+
+class GremlinServerWSProtocol(AbstractBaseProtocol):
+
+ def __init__(self, message_serializer=None, username='', password=''):
+ if message_serializer is None:
+ message_serializer = serializer.GraphSON2MessageSerializer()
+ self._message_serializer = message_serializer
+ self._username = username
+ self._password = password
+
+ def connection_made(self, transport):
+ super(GremlinServerWSProtocol, self).connection_made(transport)
+
+ def write(self, request_id, request_message):
+ message = self._message_serializer.serialize_message(
+ request_id, request_message)
+ self._transport.write(message)
+
+ def data_received(self, data, results_dict):
+ data = json.loads(data.decode('utf-8'))
+ request_id = data['requestId']
+ result_set = results_dict[request_id]
+ status_code = data['status']['code']
+ aggregate_to = data['result']['meta'].get('aggregateTo', 'list')
+ result_set._aggregate_to = aggregate_to
+ if status_code == 407:
+ auth = b''.join([b'\x00', self._username.encode('utf-8'),
+ b'\x00', self._password.encode('utf-8')])
+ request_message = request.RequestMessage(
+ 'traversal', 'authentication',
+ {'sasl': base64.b64encode(auth).decode()})
+ self.write(request_id, request_message)
+ data = self._transport.read()
+ self.data_received(data, results_dict)
+ elif status_code == 204:
+ result_set.done.set_result(None)
+ del results_dict[request_id]
+ elif status_code in [200, 206]:
+ results = []
+ for msg in data["result"]["data"]:
+ results.append(
+ self._message_serializer.deserialize_message(msg))
+ result_set.stream.put_nowait(results)
+ if status_code == 206:
+ data = self._transport.read()
+ self.data_received(data, results_dict)
+ else:
+ result_set.done.set_result(None)
+ del results_dict[request_id]
+ else:
+ result_set.stream.put_nowait(GremlinServerError(
+ "{0}: {1}".format(status_code, data["status"]["message"])))
+ result_set.done.set_result(None)
+ del results_dict[request_id]
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
index f7ed48e..a95ea10 100644
--- a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
@@ -17,11 +17,11 @@ specific language governing permissions and limitations
under the License.
'''
import abc
+import collections
import six
-from ..process.traversal import Traversal
-from ..process.traversal import TraversalStrategy
-from ..process.traversal import TraversalSideEffects
+from gremlin_python.driver import request
+from gremlin_python.process import traversal
__author__ = 'Marko A. Rodriguez (http://markorodriguez.com)'
@@ -42,45 +42,48 @@ class RemoteConnection(object):
@abc.abstractmethod
def submit(self, bytecode):
- print("sending " + bytecode + " to GremlinServer...")
- return RemoteTraversal(iter([]), TraversalSideEffects())
+ pass
def __repr__(self):
return "remoteconnection[" + self._url + "," + self._traversal_source + "]"
-class RemoteTraversal(Traversal):
+class RemoteTraversal(traversal.Traversal):
def __init__(self, traversers, side_effects):
- Traversal.__init__(self, None, None, None)
+ super(RemoteTraversal, self).__init__(None, None, None)
self.traversers = traversers
- self.side_effects = side_effects
+ self._side_effects = side_effects
+
+ @property
+ def side_effects(self):
+ return self._side_effects
-class RemoteTraversalSideEffects(TraversalSideEffects):
- def __init__(self, keys_lambda, value_lambda, close_lambda, loop):
- self._keys_lambda = keys_lambda
- self._value_lambda = value_lambda
- self._close_lambda = close_lambda
- self._loop = loop
+class RemoteTraversalSideEffects(traversal.TraversalSideEffects):
+ def __init__(self, side_effect, client):
+ self._side_effect = side_effect
+ self._client = client
self._keys = set()
self._side_effects = {}
self._closed = False
def keys(self):
- if self._loop._running:
- raise RuntimeError("Cannot call side effect methods"
- "while event loop is running")
if not self._closed:
- self._keys = self._keys_lambda()
+ message = request.RequestMessage(
+ 'traversal', 'keys',
+ {'sideEffect': self._side_effect,
+ 'aliases': {'g': self._client.traversal_source}})
+ self._keys = set(self._client.submit(message).all().result())
return self._keys
def get(self, key):
- if self._loop._running:
- raise RuntimeError("Cannot call side effect methods"
- "while event loop is running")
if not self._side_effects.get(key):
if not self._closed:
- results = self._value_lambda(key)
+ message = request.RequestMessage(
+ 'traversal', 'gather',
+ {'sideEffect': self._side_effect, 'sideEffectKey': key,
+ 'aliases': {'g': self._client.traversal_source}})
+ results = self._aggregate_results(self._client.submit(message))
self._side_effects[key] = results
self._keys.add(key)
else:
@@ -88,27 +91,59 @@ class RemoteTraversalSideEffects(TraversalSideEffects):
return self._side_effects[key]
def close(self):
- if self._loop._running:
- raise RuntimeError("Cannot call side effect methods"
- "while event loop is running")
- results = self._close_lambda()
+ if not self._closed:
+ message = request.RequestMessage(
+ 'traversal', 'close',
+ {'sideEffect': self._side_effect,
+ 'aliases': {'g': self._client._traversal_source}})
+ results = self._client.submit(message).all().result()
self._closed = True
return results
+ def _aggregate_results(self, result_set):
+ # Need to double check how all this works
+ aggregates = {'list': [], 'set': set(), 'map': {}, 'bulkset': {}}
+ results = None
+ for msg in result_set:
+ if results is None:
+ aggregate_to = result_set._aggregate_to
+ if aggregate_to:
+ results = aggregates.get(aggregate_to, [])
+ # on first message, get the right result data structure
+ # if there is no update to a structure, then the item is the result
+ # not really sure about this...
+ if results is None:
+ results = msg[0]
+ # updating a map is different than a list or a set
+ elif isinstance(results, dict):
+ if aggregate_to == "map":
+ for item in msg:
+ results.update(item)
+ else:
+ for item in msg:
+ results[item.object] = item.bulk
+ elif isinstance(results, set):
+ results.update(msg)
+ # flat add list to result list
+ else:
+ results += msg
+ if results is None:
+ results = []
+ return results
+
-class RemoteStrategy(TraversalStrategy):
+class RemoteStrategy(traversal.TraversalStrategy):
def __init__(self, remote_connection):
self.remote_connection = remote_connection
def apply(self, traversal):
if traversal.traversers is None:
remote_traversal = self.remote_connection.submit(traversal.bytecode)
+ traversal.remote_results = remote_traversal
traversal.side_effects = remote_traversal.side_effects
traversal.traversers = remote_traversal.traversers
def apply_async(self, traversal):
if traversal.traversers is None:
- remote_traversal = self.remote_connection.submit_async(
+ traversal.remote_results = self.remote_connection.submitAsync(
traversal.bytecode)
- traversal.side_effects = remote_traversal.side_effects
- traversal.traversers = remote_traversal.traversers
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/request.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/request.py b/gremlin-python/src/main/jython/gremlin_python/driver/request.py
new file mode 100644
index 0000000..0545230
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/request.py
@@ -0,0 +1,23 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+import collections
+
+
+RequestMessage = collections.namedtuple(
+ 'RequestMessage', ['processor', 'op', 'args'])
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/resultset.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/resultset.py b/gremlin-python/src/main/jython/gremlin_python/driver/resultset.py
new file mode 100644
index 0000000..e12e0a0
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/resultset.py
@@ -0,0 +1,75 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+from concurrent.futures import Future
+
+
+class ResultSet:
+
+ def __init__(self, stream, request_id):
+ self._stream = stream
+ self._request_id = request_id
+ self._done = Future()
+ self._aggregate_to = None
+
+ @property
+ def request_id(self):
+ return self._request_id
+
+ @property
+ def stream(self):
+ return self._stream
+
+ def __iter__(self):
+ return self
+
+ def __next__(self):
+ result = self.one()
+ if not result:
+ raise StopIteration
+ return result
+
+ def next(self):
+ return self.__next__()
+
+ @property
+ def done(self):
+ return self._done
+
+ def one(self):
+ if self.stream.empty() and self.done.done():
+ return
+ result = self.stream.get()
+ return result
+
+ def all(self):
+ future = Future()
+
+ def cb(f):
+ try:
+ f.result()
+ except Exception as e:
+ future.set_exception(e)
+ else:
+ results = []
+ while not self.stream.empty():
+ results += self.stream.get_nowait()
+ future.set_result(results)
+
+ self.done.add_done_callback(cb)
+ return future
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/serializer.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/serializer.py b/gremlin-python/src/main/jython/gremlin_python/driver/serializer.py
new file mode 100644
index 0000000..c39474b
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/serializer.py
@@ -0,0 +1,115 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+try:
+ import ujson as json
+except ImportError:
+ import json
+
+from gremlin_python.structure.io import graphson
+
+
+class Processor:
+ """Base class for OpProcessor serialization system."""
+
+ _graphson_writer = graphson.GraphSONWriter()
+
+ def __init__(self, default_args):
+ self._default_args = default_args
+
+ def get_op_args(self, op, args):
+ op_method = getattr(self, op, None)
+ if not op_method:
+ raise Exception("Processor does not support op: {}".format(op))
+ args_ = self._default_args.get(op, dict()).copy()
+ args_.update(args)
+ return op_method(args_)
+
+
+class GraphSON2MessageSerializer:
+ """Message serializer for GraphSONv2"""
+
+ _graphson_reader = graphson.GraphSONReader()
+
+ class traversal(Processor):
+
+ def authentication(self, args):
+ return args
+
+ def bytecode(self, args):
+ gremlin = args['gremlin']
+ args['gremlin'] = self._graphson_writer.toDict(gremlin)
+ aliases = args.get('aliases', '')
+ if not aliases:
+ aliases = {'g': 'g'}
+ args['aliases'] = aliases
+ return args
+
+ def close(self, args):
+ return self.keys(args)
+
+ def gather(self, args):
+ side_effect = args['sideEffect']
+ args['sideEffect'] = {'@type': 'g:UUID', '@value': side_effect}
+ aliases = args.get('aliases', '')
+ if not aliases:
+ aliases = {'g': 'g'}
+ args['aliases'] = aliases
+ return args
+
+ def keys(self, args):
+ side_effect = args['sideEffect']
+ args['sideEffect'] = {'@type': 'g:UUID', '@value': side_effect}
+ return args
+
+
+ def get_processor(self, processor):
+ processor = getattr(self, processor, None)
+ if not processor:
+ raise Exception("Unknown processor")
+ return processor({})
+
+ def serialize_message(self, request_id, request_message):
+ processor = request_message.processor
+ op = request_message.op
+ args = request_message.args
+ if not processor:
+ processor_obj = self.get_processor('standard')
+ else:
+ processor_obj = self.get_processor(processor)
+ args = processor_obj.get_op_args(op, args)
+ message = self.build_message(request_id, processor, op, args)
+ return message
+
+ def build_message(self, request_id, processor, op, args):
+ message = {
+ 'requestId': {'@type': 'g:UUID', '@value': request_id},
+ 'processor': processor,
+ 'op': op,
+ 'args': args
+ }
+ return self.finalize_message(message, b"\x21",
+ b"application/vnd.gremlin-v2.0+json")
+
+ def finalize_message(self, message, mime_len, mime_type):
+ message = json.dumps(message)
+ message = b''.join([mime_len, mime_type, message.encode('utf-8')])
+ return message
+
+ def deserialize_message(self, message):
+ return self._graphson_reader.toObject(message)
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/tornado/__init__.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/tornado/__init__.py b/gremlin-python/src/main/jython/gremlin_python/driver/tornado/__init__.py
new file mode 100644
index 0000000..17b49a5
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/tornado/__init__.py
@@ -0,0 +1,18 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/tornado/transport.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/tornado/transport.py b/gremlin-python/src/main/jython/gremlin_python/driver/tornado/transport.py
new file mode 100644
index 0000000..2e511b1
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/tornado/transport.py
@@ -0,0 +1,46 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+from tornado import ioloop, gen
+from tornado import websocket
+
+from gremlin_python.driver.transport import AbstractBaseTransport
+
+
+class TornadoTransport(AbstractBaseTransport):
+
+ def __init__(self):
+ self._loop = ioloop.IOLoop(make_current=False)
+
+ def connect(self, url):
+ self._ws = self._loop.run_sync(
+ lambda: websocket.websocket_connect(url))
+
+ def write(self, message):
+ self._loop.run_sync(
+ lambda: self._ws.write_message(message, binary=True))
+
+ def read(self):
+ return self._loop.run_sync(lambda: self._ws.read_message())
+
+ def close(self):
+ self._ws.close()
+ self._loop.close()
+
+ def closed(self):
+ return not self._ws.protocol
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/driver/transport.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/transport.py b/gremlin-python/src/main/jython/gremlin_python/driver/transport.py
new file mode 100644
index 0000000..453eb05
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/transport.py
@@ -0,0 +1,44 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+"""
+import abc
+import six
+
+
+@six.add_metaclass(abc.ABCMeta)
+class AbstractBaseTransport:
+
+ @abc.abstractmethod
+ def connect(self, url):
+ pass
+
+ @abc.abstractmethod
+ def write(self, message):
+ pass
+
+ @abc.abstractmethod
+ def read(self):
+ pass
+
+ @abc.abstractmethod
+ def close(self):
+ pass
+
+ @abc.abstractproperty
+ def closed(self):
+ pass
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/process/traversal.py b/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
index 6d6f2ea..e463944 100644
--- a/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
+++ b/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
@@ -26,9 +26,11 @@ class Traversal(object):
self.graph = graph
self.traversal_strategies = traversal_strategies
self.bytecode = bytecode
- self.side_effects = TraversalSideEffects()
+ self._side_effects = TraversalSideEffects()
self.traversers = None
self.last_traverser = None
+ # This is mainly to deal with futures for promise method
+ self.remote_results = None
def __repr__(self):
return str(self.bytecode)
def __eq__(self, other):
@@ -79,15 +81,16 @@ class Traversal(object):
return tempList
def promise(self, cb=None):
self.traversal_strategies.apply_async_strategies(self)
- future_traversers = self.traversers
- future = type(future_traversers)()
+ future_traversal = self.remote_results
+ future = type(future_traversal)()
def process(f):
try:
- traversers = f.result()
+ traversal = f.result()
except Exception as e:
future.set_exception(e)
else:
- self.traversers = iter(traversers)
+ self.traversers = iter(traversal.traversers)
+ self.side_effects = traversal.side_effects
if cb:
try:
result = cb(self)
@@ -97,7 +100,7 @@ class Traversal(object):
future.set_result(result)
else:
future.set_result(self)
- future_traversers.add_done_callback(process)
+ future_traversal.add_done_callback(process)
return future
@@ -407,4 +410,3 @@ class Binding(object):
return hash(self.key) + hash(self.value)
def __repr__(self):
return "binding[" + self.key + "=" + str(self.value) + "]"
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2577a13d/gremlin-python/src/main/jython/setup.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/setup.py b/gremlin-python/src/main/jython/setup.py
index 6878c34..ec345bc 100644
--- a/gremlin-python/src/main/jython/setup.py
+++ b/gremlin-python/src/main/jython/setup.py
@@ -18,8 +18,9 @@ under the License.
'''
import codecs
import os
+import sys
import time
-from setuptools import setup, Command
+from setuptools import setup
# Folder containing the setup.py
root = os.path.dirname(os.path.abspath(__file__))
@@ -43,6 +44,16 @@ from gremlin_python import __version__
version = __version__.version
+
+install_requires = [
+ 'aenum==1.4.5',
+ 'tornado==4.4.1',
+ 'six==1.10.0'
+]
+
+if sys.version_info < (3,2):
+ install_requires += ['futures==3.0.5']
+
setup(
name='gremlinpython',
version=version,
@@ -60,11 +71,7 @@ setup(
'pytest',
'mock'
],
- install_requires=[
- 'aenum==1.4.5',
- 'tornado==4.4.1',
- 'six==1.10.0'
- ],
+ install_requires=install_requires,
classifiers=[
"Intended Audience :: Developers",
"License :: OSI Approved :: Apache Software License",