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/15 15:50:57 UTC
tinkerpop git commit: added baseline driver implementation
Repository: tinkerpop
Updated Branches:
refs/heads/python_driver [created] fb7e7f255
added baseline driver implementation
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/fb7e7f25
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/fb7e7f25
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/fb7e7f25
Branch: refs/heads/python_driver
Commit: fb7e7f255585956abbb854fcc5dd3138113cf454
Parents: b0d2a27
Author: davebshow <da...@gmail.com>
Authored: Sun Jan 15 10:50:36 2017 -0500
Committer: davebshow <da...@gmail.com>
Committed: Sun Jan 15 10:50:36 2017 -0500
----------------------------------------------------------------------
.../main/jython/gremlin_python/driver/client.py | 112 +++++++++
.../jython/gremlin_python/driver/connection.py | 77 ++++++
.../driver/driver_remote_connection.py | 251 -------------------
.../jython/gremlin_python/driver/protocol.py | 100 ++++++++
.../gremlin_python/driver/remote/__init__.py | 18 ++
.../driver/remote/driver_remote_connection.py | 61 +++++
.../driver/remote/remote_connection.py | 149 +++++++++++
.../gremlin_python/driver/remote_connection.py | 114 ---------
.../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 ++++
.../gremlin_python/process/graph_traversal.py | 5 +-
.../jython/gremlin_python/process/traversal.py | 16 +-
gremlin-python/src/main/jython/setup.py | 20 +-
.../jython/tests/driver/test_threaded_driver.py | 218 ++++++++++++++++
18 files changed, 1080 insertions(+), 382 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/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..74401d4
--- /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:
+ """
+ This class provides the main interface and should be passed to
+ traversal objects using withRemote.
+ """
+ def __init__(self, url, traversal_source, protocol_factory=None,
+ transport_factory=None, pool_size=None, max_workers=None,
+ username="", password=""):
+ self._url = url
+ self._traversal_source = traversal_source
+ 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()
+ 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,
+ self._username, self._password)
+
+ 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/fb7e7f25/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..903b642
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/connection.py
@@ -0,0 +1,77 @@
+"""
+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, username, password):
+ 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._username = username
+ self._password = password
+ 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/fb7e7f25/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
deleted file mode 100644
index 2cbe0e7..0000000
--- a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
+++ /dev/null
@@ -1,251 +0,0 @@
-"""
-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 base64
-import functools
-import json
-import uuid
-from tornado import gen
-from tornado import concurrent
-from tornado import ioloop
-from tornado import websocket
-
-from gremlin_python.structure.io.graphson import GraphSONReader, GraphSONWriter
-from .remote_connection import RemoteConnection
-from .remote_connection import RemoteTraversal
-from .remote_connection import RemoteTraversalSideEffects
-
-
-class GremlinServerError(Exception):
- pass
-
-
-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 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-v2.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
- 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")
-
- # authentification required then
- if status_code == 407:
- self._websocket.write_message(
- b"".join([b"\x21",
- b"application/vnd.gremlin-v2.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"]))
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/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..70404d4
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/protocol.py
@@ -0,0 +1,100 @@
+"""
+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 collections
+import json
+import uuid
+
+import six
+
+from gremlin_python.driver import serializer
+
+
+@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):
+ if message_serializer is None:
+ message_serializer = serializer.GraphSON2MessageSerializer()
+ self._message_serializer = message_serializer
+
+ 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:
+ self._authenticate(request_id)
+ data = self._transport.read()
+ self.receive(data, response_queues)
+ elif status_code == 204:
+ # result_set.stream.put_nowait([None])
+ 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.receive(data, response_queues)
+ 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]
+
+ def _authenticate(self, request_id):
+ auth = b''.join([b'\x00', self.username.encode('utf-8'),
+ b'\x00', self.password.encode('utf-8')])
+ args = {'sasl': base64.b64encode(auth).decode(),
+ 'saslMechanism': 'PLAIN'}
+ message = self._message_serializer.serialize_message(
+ request_id, '', 'authentication', **args)
+ self._transport.write(message)
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/gremlin-python/src/main/jython/gremlin_python/driver/remote/__init__.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/remote/__init__.py b/gremlin-python/src/main/jython/gremlin_python/driver/remote/__init__.py
new file mode 100644
index 0000000..17b49a5
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/remote/__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/fb7e7f25/gremlin-python/src/main/jython/gremlin_python/driver/remote/driver_remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/remote/driver_remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/remote/driver_remote_connection.py
new file mode 100644
index 0000000..5e949a1
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/remote/driver_remote_connection.py
@@ -0,0 +1,61 @@
+"""
+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
+
+from gremlin_python.driver import client
+from gremlin_python.driver.remote.remote_connection import (
+ RemoteTraversal, RemoteTraversalSideEffects)
+
+
+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,
+ username, password)
+
+ def close(self):
+ self._client.close()
+
+ def submit(self, bytecode):
+ 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 = result_set.all().result()
+ side_effects = RemoteTraversalSideEffects(result_set.request_id,
+ self._client)
+ future.set_result(RemoteTraversal(iter(results), side_effects))
+
+ future_result_set.add_done_callback(cb)
+ return future
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/gremlin-python/src/main/jython/gremlin_python/driver/remote/remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/remote/remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/remote/remote_connection.py
new file mode 100644
index 0000000..a95ea10
--- /dev/null
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/remote/remote_connection.py
@@ -0,0 +1,149 @@
+'''
+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 collections
+import six
+
+from gremlin_python.driver import request
+from gremlin_python.process import traversal
+
+__author__ = 'Marko A. Rodriguez (http://markorodriguez.com)'
+
+
+@six.add_metaclass(abc.ABCMeta)
+class RemoteConnection(object):
+ def __init__(self, url, traversal_source):
+ self._url = url
+ self._traversal_source = traversal_source
+
+ @property
+ def url(self):
+ return self._url
+
+ @property
+ def traversal_source(self):
+ return self._traversal_source
+
+ @abc.abstractmethod
+ def submit(self, bytecode):
+ pass
+
+ def __repr__(self):
+ return "remoteconnection[" + self._url + "," + self._traversal_source + "]"
+
+
+class RemoteTraversal(traversal.Traversal):
+ def __init__(self, traversers, side_effects):
+ super(RemoteTraversal, self).__init__(None, None, None)
+ self.traversers = traversers
+ self._side_effects = side_effects
+
+ @property
+ def side_effects(self):
+ return self._side_effects
+
+
+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 not self._closed:
+ 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 not self._side_effects.get(key):
+ if not self._closed:
+ 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:
+ return None
+ return self._side_effects[key]
+
+ def close(self):
+ 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(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:
+ traversal.remote_results = self.remote_connection.submitAsync(
+ traversal.bytecode)
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/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
deleted file mode 100644
index f7ed48e..0000000
--- a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
+++ /dev/null
@@ -1,114 +0,0 @@
-'''
-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
-
-from ..process.traversal import Traversal
-from ..process.traversal import TraversalStrategy
-from ..process.traversal import TraversalSideEffects
-
-__author__ = 'Marko A. Rodriguez (http://markorodriguez.com)'
-
-
-@six.add_metaclass(abc.ABCMeta)
-class RemoteConnection(object):
- def __init__(self, url, traversal_source):
- self._url = url
- self._traversal_source = traversal_source
-
- @property
- def url(self):
- return self._url
-
- @property
- def traversal_source(self):
- return self._traversal_source
-
- @abc.abstractmethod
- def submit(self, bytecode):
- print("sending " + bytecode + " to GremlinServer...")
- return RemoteTraversal(iter([]), TraversalSideEffects())
-
- def __repr__(self):
- return "remoteconnection[" + self._url + "," + self._traversal_source + "]"
-
-
-class RemoteTraversal(Traversal):
- def __init__(self, traversers, side_effects):
- Traversal.__init__(self, None, None, None)
- self.traversers = traversers
- self.side_effects = 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
- 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()
- 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)
- self._side_effects[key] = results
- self._keys.add(key)
- else:
- return None
- 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()
- self._closed = True
- return results
-
-
-class RemoteStrategy(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.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.bytecode)
- traversal.side_effects = remote_traversal.side_effects
- traversal.traversers = remote_traversal.traversers
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/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/fb7e7f25/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/fb7e7f25/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/fb7e7f25/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/fb7e7f25/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/fb7e7f25/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/fb7e7f25/gremlin-python/src/main/jython/gremlin_python/process/graph_traversal.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/process/graph_traversal.py b/gremlin-python/src/main/jython/gremlin_python/process/graph_traversal.py
index c07853a..7a847d3 100644
--- a/gremlin-python/src/main/jython/gremlin_python/process/graph_traversal.py
+++ b/gremlin-python/src/main/jython/gremlin_python/process/graph_traversal.py
@@ -21,7 +21,7 @@ from .traversal import Traversal
from .traversal import TraversalStrategies
from .strategies import VertexProgramStrategy
from .traversal import Bytecode
-from ..driver.remote_connection import RemoteStrategy
+from gremlin_python.driver.remote.remote_connection import RemoteStrategy
from .. import statics
from ..statics import long
@@ -1146,6 +1146,3 @@ def where(*args):
return __.where(*args)
statics.add_static('where', where)
-
-
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/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/fb7e7f25/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 3c55554..f0a6961 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,17 @@ 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 +72,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",
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/fb7e7f25/gremlin-python/src/main/jython/tests/driver/test_threaded_driver.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/tests/driver/test_threaded_driver.py b/gremlin-python/src/main/jython/tests/driver/test_threaded_driver.py
new file mode 100644
index 0000000..6450c5e
--- /dev/null
+++ b/gremlin-python/src/main/jython/tests/driver/test_threaded_driver.py
@@ -0,0 +1,218 @@
+import sys
+import concurrent.futures
+import uuid
+from threading import Thread
+
+import pytest
+from six.moves import queue
+from tornado import ioloop, gen
+
+from gremlin_python.driver.client import Client
+from gremlin_python.driver.request import RequestMessage
+from gremlin_python.driver.connection import Connection
+from gremlin_python.driver.protocol import GremlinServerWSProtocol
+from gremlin_python.driver.remote.driver_remote_connection import (
+ DriverRemoteConnection)
+from gremlin_python.driver.tornado.transport import TornadoTransport
+from gremlin_python.structure.graph import Graph
+
+
+def test_connection():
+ protocol = GremlinServerWSProtocol()
+ executor = concurrent.futures.ThreadPoolExecutor(5)
+ pool = queue.Queue()
+ conn = Connection('ws://localhost:8182/gremlin', 'g', protocol,
+ lambda: TornadoTransport(), executor, pool, '', '')
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode})
+ results_set = conn.write(message).result(1)
+ future = results_set.all()
+ results = future.result()
+ assert len(results) == 6
+ assert isinstance(results, list)
+ conn.close()
+ executor.shutdown()
+
+def test_client():
+ client = Client('ws://localhost:8182/gremlin', 'g')
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode})
+ result_set = client.submit(message)
+ assert len(result_set.all().result()) == 6
+ client.close()
+
+def test_iterate_result_set():
+ client = Client('ws://localhost:8182/gremlin', 'g')
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode})
+ result_set = client.submit(message)
+ results = []
+ for result in result_set:
+ results += result
+ assert len(results) == 6
+ client.close()
+
+def test_client_async():
+ client = Client('ws://localhost:8182/gremlin', 'g')
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode})
+ future = client.submitAsync(message)
+ assert not future.done()
+ result_set = future.result()
+ assert len(result_set.all().result()) == 6
+ client.close()
+
+def test_connection_share():
+ client = Client('ws://localhost:8182/gremlin', 'g', pool_size=1)
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode})
+ future = client.submitAsync(message)
+ future2 = client.submitAsync(message)
+
+ result_set2 = future2.result()
+ assert len(result_set2.all().result()) == 6
+
+ # This future has to finish for the second to yield result - pool_size=1
+ assert future.done()
+ result_set = future.result()
+ assert len(result_set.all().result()) == 6
+ client.close()
+
+def test_multi_conn_pool():
+ client = Client('ws://localhost:8182/gremlin', 'g', pool_size=2)
+ g = Graph().traversal()
+ t = g.V()
+ message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode})
+ future = client.submitAsync(message)
+ future2 = client.submitAsync(message)
+
+ result_set2 = future2.result()
+ assert len(result_set2.all().result()) == 6
+
+ # with connection pool `future` may or may not be done here
+ result_set = future.result()
+ assert len(result_set.all().result()) == 6
+ client.close()
+
+def test_driver_remote_conn():
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ g = Graph().traversal().withRemote(conn)
+ t = g.V()
+ assert len(t.toList()) == 6
+ conn.close()
+
+def test_promise():
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ g = Graph().traversal().withRemote(conn)
+ future = g.V().promise()
+ t = future.result()
+ assert len(t.toList()) == 6
+ conn.close()
+
+def test_clients_in_threads_with_promise():
+ q = queue.Queue()
+ child = Thread(target=_executor, args=(q, None))
+ child2 = Thread(target=_executor, args=(q, None))
+ child.start()
+ child2.start()
+ for x in range(2):
+ success = q.get()
+ assert success == 'success!'
+ child.join()
+ child2.join()
+
+def test_client_in_threads_with_promise():
+ q = queue.Queue()
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ child = Thread(target=_executor, args=(q, conn))
+ child2 = Thread(target=_executor, args=(q, conn))
+ child.start()
+ child2.start()
+ for x in range(2):
+ success = q.get()
+ assert success == 'success!'
+ child.join()
+ child2.join()
+ conn.close()
+
+def _executor(q, conn):
+ if not conn:
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ try:
+ g = Graph().traversal().withRemote(conn)
+ future = g.V().promise()
+ t = future.result()
+ assert len(t.toList()) == 6
+ except:
+ q.put(sys.exc_info()[0])
+ else:
+ q.put('success!')
+ conn.close()
+
+def test_in_tornado_app():
+
+ @gen.coroutine
+ def go():
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ g = Graph().traversal().withRemote(conn)
+ yield gen.sleep(0)
+ assert len(g.V().toList()) == 6
+ conn.close()
+
+ io_loop = ioloop.IOLoop.current()
+ io_loop.run_sync(go)
+
+
+def test_side_effects():
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ g = Graph().traversal().withRemote(conn)
+ t = g.V().aggregate('a').aggregate('b')
+ t.toList()
+
+ # The 'a' key should return some side effects
+ results = t.side_effects.get('a')
+ assert results
+
+ # Close result is None
+ results = t.side_effects.close()
+ assert not results
+
+ # Shouldn't get any new info from server
+ # 'b' isn't in local cache
+ results = t.side_effects.get('b')
+ assert not results
+
+ # But 'a' should still be cached locally
+ results = t.side_effects.get('a')
+ assert results
+
+ # 'a' should have been added to local keys cache, but not 'b'
+ results = t.side_effects.keys()
+ assert len(results) == 1
+ a, = results
+ assert a == 'a'
+
+ # Try to get 'b' directly from server, should throw error
+ with pytest.raises(Exception):
+ t.side_effects.get('b')
+ connection.close()
+
+def test_side_effect_keys():
+ conn = DriverRemoteConnection(
+ 'ws://localhost:8182/gremlin', 'g', pool_size=4)
+ g = Graph().traversal().withRemote(conn)
+ t = g.V().aggregate('a')
+ t.toList()
+ result, = t.side_effects.keys()
+ assert result == 'a'