Repository: dieseldev/diesel (language: python)

Issue:
Port to Python3
Hi there,

Diesel seems a really interesting project, such a shame it is not available for python3 !
So I've forked the repo (https://github.com/touilleMan/diesel/tree/py3k) and started to port the code ;-)

So far, all the tests are passing for Python 2.6, 2.7, 3.3 and 3.4 using the same codebase, but it seems those tests are not really exhaustive and there is still some work to do !

Now I'm trying to make work all the examples available.

My main point of concern is how to handle the strings:
- in Python2, `str` is an array of bytes (`bytes` type is an alias of `str`), to use unicode we must is `unicode` type
- in Python3, `str` is an array of unicodes characters and only `bytes` an array of bytes

Considering Diesel, all the code is using classical `str` types (i.e. array of bytes), this seems pretty logical to me given there is no standard way to determine the encoding of data coming from a socket.

The downside of this is to make the code compatible with Python3, all the strings needs to be explicitly converted to bytes, leading to a lot of small changes, an code more complex (numerous conversions unicode => bytes).

Example:

``` python
# Python2 only
send('AUTH', self.password)
send("HTTP/%s %s %s\r\n" % (('%s.%s' % version), resp.status_code, resp.status))
# Python2&3
send(b'AUTH', self.password.encode())
send(("HTTP/%s %s %s\r\n" % (('%s.%s' % version), resp.status_code, resp.status)).encode())
```

I've started to see if the core.send/core.receive function could be used to handle encoding, but - although it would be a nice solution - I don't believe this could lead to a easy drop-in replacement.

``` python
def send(data, priority=5, encoding='ascii'):
    if isinstance(data, builtins.str):
        data = data.encode(encoding)
    return current_loop.send(data, priority=priority)

def receive(spec=None, encoding='ascii'):
    data = current_loop.input_op(spec)
    if isinstance(data, builtins.bytes):
        data = data.decode(encoding)
    return data
```

Do you have any advice/idea about the best way to handle this ?



Discussion / hints:

Fix patch:
diff --git a/.travis.yml b/.travis.yml
index 51a71d3..2bfe23c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,8 +1,13 @@
 language: python
 python:
+  - "3.4"
+  - "3.3"
   - "2.7"
   - "2.6"
+services:
+  - redis-server
+  - mongodb
 # command to install dependencies, e.g. pip install -r requirements.txt --use-mirrors
 install: python setup.py install
 # command to run tests, e.g. python setup.py test
-script:  dnosetests tests/unit tests/integration
+script:  dnosetests tests/unit tests/integration tests/protocol
diff --git a/README.rst b/README.rst
index 259dfa4..046d84e 100644
--- a/README.rst
+++ b/README.rst
@@ -1,3 +1,5 @@
+.. image:: https://travis-ci.org/touilleMan/diesel.svg?branch=py3k
+    :target: https://travis-ci.org/touilleMan/diesel
 
 Why Diesel?
 ===========
diff --git a/diesel/__init__.py b/diesel/__init__.py
index 2cbb383..1b58b2b 100644
--- a/diesel/__init__.py
+++ b/diesel/__init__.py
@@ -1,6 +1,6 @@
 # vim:ts=4:sw=4:expandtab
 from .logmod import log, levels as loglevels, set_log_level
-import events
+from . import events
 from .core import sleep, Loop, wait, fire, thread, until, signal, datagram
 from .core import until_eol, send, receive, first, fork, fork_child, label, fork_from_thread
 from .core import ParentDiedException, TerminateLoop
diff --git a/diesel/app.py b/diesel/app.py
index 89069e4..e1e8be1 100644
--- a/diesel/app.py
+++ b/diesel/app.py
@@ -5,7 +5,7 @@
 import gc
 import cProfile
 import traceback
-
+import collections
 from greenlet import greenlet
 
 from diesel import runtime
@@ -74,7 +74,7 @@ def _main():
                 except KeyboardInterrupt:
                     log.warning("-- KeyboardInterrupt raised.. exiting main loop --")
                     break
-                except Exception, e:
+                except Exception as e:
                     log.error("-- Unhandled Exception rose to main loop --")
                     log.error(traceback.format_exc())
 
@@ -93,12 +93,11 @@ def _profiled_main():
                 config['filename'] = statsfile
             try:
                 cProfile.runctx('_real_main()', globals(), locals(), **config)
-            except TypeError, e:
+            except TypeError as e:
                 if "sort" in e.args[0]:
                     del config['sort']
                     cProfile.runctx('_real_main()', globals(), locals(), **config)
                 else: raise e
-
         self.runhub = greenlet(_main if not profile else _profiled_main)
         self.runhub.switch()
 
@@ -173,7 +172,7 @@ def quickstart(*args, **kw):
             app.add_service(a)
         elif isinstance(a, Loop):
             app.add_loop(a)
-        elif callable(a):
+        elif isinstance(a, collections.Callable):
             app.add_loop(Loop(a))
     app.run()
 
diff --git a/diesel/buffer.py b/diesel/buffer.py
index a6c7fb0..6e3b379 100644
--- a/diesel/buffer.py
+++ b/diesel/buffer.py
@@ -1,3 +1,5 @@
+from builtins import int, object
+
 # vim:ts=4:sw=4:expandtab
 class BufAny(object):
     pass
@@ -17,9 +19,8 @@ def __init__(self):
     def set_term(self, term):
         '''Set the current sentinel.
 
-        `term` is either an int, for a byte count, or
-        a string, for a sequence of characters that needs
-        to occur in the byte stream.
+        :param term: Either an int, for a bytes count, or bytes, for a
+        sequence of characters that needs to occur in the byte stream.
         '''
         self._atterm = term
 
@@ -46,13 +47,13 @@ def check(self):
             if self.has_data:
                 return self.pop()
             return None
-        if type(self._atterm) is int:
+        if isinstance(self._atterm, int):
             if self._atmark >= self._atterm:
                 ind = self._atterm
         elif self._atterm is None:
             return None
         else:
-            all = ''.join(self._atinbuf)
+            all = b''.join(self._atinbuf)
             res = all.find(self._atterm)
             if res != -1:
                 ind = res + len(self._atterm)
@@ -60,7 +61,7 @@ def check(self):
             return None
         self._atterm = None # this terminator was used
         if all is None:
-            all = ''.join(self._atinbuf)
+            all = b''.join(self._atinbuf)
         use = all[:ind]
         new_all = all[ind:]
         self._atinbuf = [new_all]
@@ -69,7 +70,7 @@ def check(self):
         return use
 
     def pop(self):
-        b = ''.join(self._atinbuf)
+        b = b''.join(self._atinbuf)
         self._atinbuf = []
         self._atmark = 0
         return b
diff --git a/diesel/console.py b/diesel/console.py
index 136a97a..97fea2a 100644
--- a/diesel/console.py
+++ b/diesel/console.py
@@ -31,7 +31,7 @@
 import struct
 import sys
 
-from cStringIO import StringIO
+from io import StringIO
 
 import diesel
 
@@ -70,7 +70,7 @@ def runcode(self, ignored_codeobj):
             (sz,) = struct.unpack('>Q', header)
             if sz:
                 data = receive(sz)
-                print data.rstrip()
+                print(data.rstrip())
 
 def console_for(pid):
     """Sends a SIGTRAP to the pid and returns a console UI handler.
@@ -170,7 +170,7 @@ def main():
         parser.print_usage()
         raise SystemExit(1)
     if args[0] == 'dummy':
-        print "PID", os.getpid()
+        print("PID", os.getpid())
         def wait_for_signal():
             log_ = log.name('dummy')
             log_.min_level = loglevels.INFO
diff --git a/diesel/core.py b/diesel/core.py
index 51f2f3f..8e9143b 100644
--- a/diesel/core.py
+++ b/diesel/core.py
@@ -5,11 +5,13 @@
 import os
 import itertools
 from greenlet import greenlet
+from past.builtins import basestring
 
 from diesel import buffer
 from diesel import runtime
 from diesel.logmod import log
 from diesel.events import EarlyValue
+import collections
 
 
 class LoopKeepAlive(Exception):
@@ -17,17 +19,27 @@ class LoopKeepAlive(Exception):
     allows the app to re-schedule keep_alive loops.
     '''
 
+
 class ParentDiedException(Exception):
     '''Raised when the parent (assigned via fork_child) has died.
     '''
 
+
 class TerminateLoop(Exception):
     '''Raised to terminate the current loop, closing the socket if there
     is one associated with the loop.
     '''
 
-CRLF = '\r\n'
+
+CRLF = b'\r\n'
 BUFSIZ = 2 ** 14
+TOK_RECEIVE_ANY = 'receive_any'
+TOK_RECEIVE = 'receive'
+TOK_UNTIL = 'until'
+TOK_UNTIL_EOL = 'until_eol'
+TOK_DATAGRAM = 'datagram'
+TOK_SLEEP = 'sleep'
+
 
 def until(sentinel):
     """Returns data from the underlying connection, terminated by sentinel.
@@ -37,12 +49,13 @@ def until(sentinel):
     off the socket beyond the sentinel is buffered.
 
     :param sentinel: The sentinel to wait for before returning data.
-    :type sentinel: A byte string (str).
-    :return: A byte string (str).
+    :type sentinel: bytes
+    :return: bytes
 
     """
     return _current_loop.input_op(sentinel)
 
+
 def until_eol():
     """Returns data from the underlying connection, terminated by \\r\\n.
 
@@ -50,7 +63,7 @@ def until_eol():
     a carriage return and a line feed (CRLF). Data that has been read off the
     socket beyond the CRLF is buffered.
 
-    :return: A byte string (str).
+    :return: bytes
 
     """
     return until(CRLF)
@@ -75,50 +88,65 @@ def receive(spec=None):
     """
     return _current_loop.input_op(spec)
 
+
 def send(data, priority=5):
     """Sends data out over the underlying connection.
 
     :param data: The data that you want to send.
-    :type data: A byte string (str).
+    :type data: bytes or bytes-stream (i.g. file).
     :param priority: The priority
 
     """
+    if isinstance(data, basestring) and not isinstance(data, bytes):
+        raise RuntimeError("Can only send bytes, you should use "
+                           "`encode` prior to call this function")
     return _current_loop.send(data, priority=priority)
 
+
 def wait(*args, **kw):
     return _current_loop.wait(*args, **kw)
 
+
 def fire(*args, **kw):
     return _current_loop.fire(*args, **kw)
 
+
 def sleep(*args, **kw):
     return _current_loop.sleep(*args, **kw)
 
+
 def thread(*args, **kw):
     return _current_loop.thread(*args, **kw)
 
+
 def first(*args, **kw):
     return _current_loop.first(*args, **kw)
 
+
 def label(*args, **kw):
     return _current_loop.label(*args, **kw)
 
+
 def fork(*args, **kw):
     return _current_loop.fork(False, *args, **kw)
 
+
 def fork_child(*args, **kw):
     return _current_loop.fork(True, *args, **kw)
 
+
 def fork_from_thread(f, *args, **kw):
     l = Loop(f, *args, **kw)
     runtime.current_app.hub.schedule_loop_from_other_thread(l, ContinueNothing)
 
+
 def signal(sig, callback=None):
     if not callback:
         return _current_loop.signal(sig)
     else:
         return _current_loop._signal(sig, callback)
 
+
 class datagram(object):
     """Used to create a singleton instance of the same name.
 
@@ -126,17 +154,22 @@ class datagram(object):
 
     """
     pass
+
+
 datagram = datagram()
 _datagram = datagram
-
 _current_loop = None
 
+
 ContinueNothing = object()
 
+
 def identity(cb): return cb
 
+
 ids = itertools.count(1)
 
+
 class Loop(object):
     def __init__(self, loop_callable, *args, **kw):
         self.loop_callable = loop_callable
@@ -146,7 +179,7 @@ def __init__(self, loop_callable, *args, **kw):
         self.keep_alive = False
         self.hub = runtime.current_app.hub
         self.app = runtime.current_app
-        self.id = ids.next()
+        self.id = next(ids)
         self.children = set()
         self.parent = None
         self.deaths = 0
@@ -181,7 +214,8 @@ def run(self):
             raise
         except ParentDiedException:
             parent_died = True
-        except:
+        except Exception as e:
+            print(e)
             log.trace().error("-- Unhandled Exception in local loop <%s> --" % self.loop_label)
         finally:
             if self.connection_stack:
@@ -268,25 +302,25 @@ def mark(d):
                 return mark
             return deco
 
-        f_sent = filter(None, (receive_any, receive, until, until_eol, datagram))
-        assert len(f_sent) <= 1,(
-        "only 1 of (receive_any, receive, until, until_eol, datagram) may be provided")
+        f_sent = [_f for _f in (receive_any, receive, until, until_eol, datagram) if _f]
+        assert len(f_sent) <= 1, ("only 1 of (receive_any, receive, until,"
+                                  " until_eol, datagram) may be provided")
         sentinel = None
         if receive_any:
             sentinel = buffer.BufAny
-            tok = 'receive_any'
+            tok = TOK_RECEIVE_ANY
         elif receive:
             sentinel = receive
-            tok = 'receive'
+            tok = TOK_RECEIVE
         elif until:
             sentinel = until
-            tok = 'until'
+            tok = TOK_UNTIL
         elif until_eol:
-            sentinel = "\r\n"
-            tok = 'until_eol'
+            sentinel = CRLF
+            tok = TOK_UNTIL_EOL
         elif datagram:
             sentinel = _datagram
-            tok = 'datagram'
+            tok = TOK_DATAGRAM
         if sentinel:
             early_val = self._input_op(sentinel, marked_cb(tok))
             if early_val:
@@ -294,7 +328,7 @@ def mark(d):
             # othewise.. process others and dispatch
 
         if sleep is not None:
-            self._sleep(sleep, marked_cb('sleep'))
+            self._sleep(sleep, marked_cb(TOK_SLEEP))
 
         if waits:
             for w in waits:
@@ -327,7 +361,7 @@ def fire_in(self, what, value):
 
     def wait(self, event):
         v = self._wait(event)
-        if type(v) is EarlyValue:
+        if isinstance(v, EarlyValue):
             self.reschedule_with_this_value(v.val)
         return self.dispatch()
 
@@ -338,7 +372,7 @@ def call_in():
                 rcb(d)
             self.hub.schedule(call_in)
         v = self.app.waits.wait(self, event)
-        if type(v) is EarlyValue:
+        if isinstance(v, EarlyValue):
             return v
         self.fire_handlers[v] = cb
 
@@ -414,7 +448,7 @@ def _input_op(self, sentinel, cb_maker=identity):
         conn = self.check_connection()
         cb = cb_maker(self.wake)
         res = conn.check_incoming(sentinel, cb)
-        if callable(res):
+        if isinstance(res, collections.Callable):
             cb = res
         elif res:
             return res
@@ -446,4 +480,3 @@ def signal(self, sig):
 
     def _signal(self, sig, cb):
         self.hub.add_signal_handler(sig, cb)
-
diff --git a/diesel/events.py b/diesel/events.py
index 4d6e1e5..785624d 100644
--- a/diesel/events.py
+++ b/diesel/events.py
@@ -25,7 +25,6 @@ class StringWaiter(str, Waiter):
     def wait_id(self):
         return str(self)
 
-
 class WaitPool(object):
     '''A structure that manages all `wait`ers, makes sure fired events
     get to the right places.
@@ -35,7 +34,7 @@ def __init__(self):
         self.loop_refs = defaultdict(set)
 
     def wait(self, who, what):
-        if isinstance(what, basestring):
+        if isinstance(what, str):
             what = StringWaiter(what)
 
         if what.ready_early():
@@ -46,7 +45,7 @@ def wait(self, who, what):
         return what.wait_id
 
     def fire(self, what, value):
-        if isinstance(what, basestring):
+        if isinstance(what, str):
             what = StringWaiter(what)
 
         static = False
@@ -58,7 +57,7 @@ def fire(self, what, value):
                     value = what.process_fire(value)
                 except StopWaitDispatch:
                     break
-                if type(value) == StaticValue:
+                if isinstance(value, StaticValue):
                     static = True
                     value = value.value
             handler.fire_in(what.wait_id, value)
diff --git a/diesel/hub.py b/diesel/hub.py
index c9545ef..4319575 100644
--- a/diesel/hub.py
+++ b/diesel/hub.py
@@ -18,12 +18,18 @@
 import fcntl
 import os
 import signal
-import thread
+try:
+    import _thread
+except ImportError:
+    import thread as _thread
 
 from collections import deque, defaultdict
 from operator import attrgetter
 from time import time
-from Queue import Queue, Empty
+try:
+    from queue import Queue, Empty
+except ImportError:
+    from Queue import Queue, Empty
 
 TRIGGER_COMPARE = attrgetter('trigger_time')
 
@@ -121,16 +127,16 @@ def run_in_thread(self, reschedule, f, *args, **kw):
         def wrap():
             try:
                 res = f(*args, **kw)
-            except Exception, e:
+            except Exception as e:
                 self.thread_comp_in.put((reschedule, e))
             else:
                 self.thread_comp_in.put((reschedule, res))
             self.wake_from_other_thread()
-        thread.start_new_thread(wrap, ())
+        _thread.start_new_thread(wrap, ())
 
     def wake_from_other_thread(self):
         try:
-            os.write(self._t_wakeup, '\0')
+            os.write(self._t_wakeup, b'\0')
         except IOError:
             pass
 
@@ -279,7 +285,7 @@ def handle_events(self):
 
                 if not self.run:
                     return
-        except IOError, e:
+        except IOError as e:
             if e.errno == errno.EINTR:
                 while self.run_now and self.run:
                     self.run_now.popleft()()
diff --git a/diesel/interactive.py b/diesel/interactive.py
index a95f1f4..5cdb531 100644
--- a/diesel/interactive.py
+++ b/diesel/interactive.py
@@ -8,6 +8,7 @@
 installed).
 
 """
+from __future__ import print_function
 import code
 import sys
 sys.path.insert(0, '.')
@@ -63,7 +64,7 @@ def python():
 
 def ipython():
     if not IPYTHON_AVAILABLE:
-        print >> sys.stderr, "IPython not found."
+        print("IPython not found.", file=sys.stderr)
         raise SystemExit(1)
     quickstart(interact_ipython)
 
diff --git a/diesel/logmod.py b/diesel/logmod.py
index 37a4419..869ae65 100644
--- a/diesel/logmod.py
+++ b/diesel/logmod.py
@@ -13,7 +13,6 @@
 
 diesel_format = formats.line_format
 diesel_format.traceback_prefix = '\n'
-diesel_format.conversion = formats.ConversionTable()
 diesel_format.conversion.add("time", partial(time.strftime, "%Y/%m/%d %H:%M:%S"), "[{1}]".format)
 diesel_format.conversion.add("name", str, "{{{1}}}".format)
 diesel_format.conversion.add("level", str, "{1}".format)
diff --git a/diesel/pipeline.py b/diesel/pipeline.py
index 7920c83..b2b09d5 100644
--- a/diesel/pipeline.py
+++ b/diesel/pipeline.py
@@ -2,17 +2,47 @@
 '''An outgoing pipeline that can handle
 strings or files.
 '''
+import sys
 try:
-    import cStringIO
+    import io
 except ImportError:
-    raise ImportError, "cStringIO is required"
+    raise ImportError("cStringIO is required")
 
+from builtins import object, bytes, str
 from bisect import bisect_right
+# functools.total_ordering is not available for Python < 2.7
+try:
+    from functools import total_ordering
+except ImportError:
+    def total_ordering(cls):
+        """Class decorator that fills in missing ordering methods"""
+        convert = {
+            '__lt__': [('__gt__', lambda self, other: not (self < other or self == other)),
+                       ('__le__', lambda self, other: self < other or self == other),
+                       ('__ge__', lambda self, other: not self < other)],
+            '__le__': [('__ge__', lambda self, other: not self <= other or self == other),
+                       ('__lt__', lambda self, other: self <= other and not self == other),
+                       ('__gt__', lambda self, other: not self <= other)],
+            '__gt__': [('__lt__', lambda self, other: not (self > other or self == other)),
+                       ('__ge__', lambda self, other: self > other or self == other),
+                       ('__le__', lambda self, other: not self > other)],
+            '__ge__': [('__le__', lambda self, other: (not self >= other) or self == other),
+                       ('__gt__', lambda self, other: self >= other and not self == other),
+                       ('__lt__', lambda self, other: not self >= other)]
+        }
+        roots = set(dir(cls)) & set(convert)
+        if not roots:
+            raise ValueError('must define at least one ordering operation: < > <= >=')
+        root = max(roots)       # prefer __lt__ to __le__ to __gt__ to __ge__
+        for opname, opfunc in convert[root]:
+            if opname not in roots:
+                opfunc.__name__ = opname
+                opfunc.__doc__ = getattr(int, opname).__doc__
+                setattr(cls, opname, opfunc)
+        return cls
 
-_obj_SIO = cStringIO.StringIO
-_type_SIO = cStringIO.OutputType
 def make_SIO(d):
-    t = _obj_SIO()
+    t = io.BytesIO()
     t.write(d)
     t.seek(0)
     return t
@@ -27,9 +57,14 @@ def get_file_length(f):
 class PipelineCloseRequest(Exception): pass
 class PipelineClosed(Exception): pass
 
+@total_ordering
 class PipelineItem(object):
     def __init__(self, d):
-        if type(d) is str:
+        # TODO : error on str
+        if isinstance(d, str):
+            d = d.encode()
+            import pdb; pdb.set_trace()
+        if isinstance(d, bytes):
             self.f = make_SIO(d)
             self.length = len(d)
             self.is_sio = True
@@ -39,7 +74,7 @@ def __init__(self, d):
             self.length = get_file_length(d)
             self.is_sio = False
         else:
-            raise ValueError("argument to add() must be either a str or a file-like object")
+            raise ValueError("argument to add() must be either bytes or a file-like object")
         self.read = self.f.read
 
     def merge(self, s):
@@ -55,15 +90,15 @@ def reset(self):
     def done(self):
         return self.f.tell() == self.length
 
-    def __cmp__(self, other):
+    def __lt__(self, other):
         if other is PipelineStandIn:
-            return -1
-        return cmp(self, other) 
+            return True
+        return self < other
 
 class PipelineStandIn(object): pass
 
 class Pipeline(object):
-    '''A pipeline that supports appending strings or
+    '''A pipeline that supports appending bytes or
     files and can read() transparently across object
     boundaries in the outgoing buffer.
     '''
@@ -82,7 +117,7 @@ def add(self, d, priority=5):
 
         dummy = (priority, PipelineStandIn)
         ind = bisect_right(self.line, dummy)
-        if ind > 0 and type(d) is str and self.line[ind - 1][-1].is_sio:
+        if ind > 0 and isinstance(d, bytes) and self.line[ind - 1][-1].is_sio:
             a_pri, adjacent = self.line[ind - 1]
             if adjacent.is_sio and a_pri == priority:
                 adjacent.merge(d)
@@ -108,19 +143,19 @@ def read(self, amt):
         if not self.current and not self.line:
             if self.want_close:
                 raise PipelineCloseRequest()
-            return ''
+            return b''
 
         if not self.current:
             _, self.current = self.line.pop(0)
             self.current.reset()
 
-        out = ''
+        out = b''
         while len(out) < amt:
             try:
                 data = self.current.read(amt - len(out))
             except ValueError:
-                data = ''
-            if data == '':
+                data = b''
+            if not data:
                 if not self.line:
                     self.current = None
                     break
diff --git a/diesel/protocols/DNS.py b/diesel/protocols/DNS.py
index 2ea65f0..ef82ac5 100644
--- a/diesel/protocols/DNS.py
+++ b/diesel/protocols/DNS.py
@@ -1,4 +1,4 @@
-from __future__ import absolute_import
+
 
 import time
 from collections import deque
@@ -23,7 +23,7 @@ class Timeout(Exception):
 _search_domains = []
 if _resolv_conf.domain:
     _search_domains.append(str(_resolv_conf.domain)[:-1])
-_search_domains.extend(map(lambda n: str(n)[:-1], _resolv_conf.search))
+_search_domains.extend([str(n)[:-1] for n in _resolv_conf.search])
 
 del _resolv_conf
 
diff --git a/diesel/protocols/http/core.py b/diesel/protocols/http/core.py
index c7d13b2..71dd997 100644
--- a/diesel/protocols/http/core.py
+++ b/diesel/protocols/http/core.py
@@ -2,14 +2,17 @@
 '''HTTP/1.1 implementation of client and server.
 '''
 
-import cStringIO
+from future.standard_library import install_aliases
+install_aliases()
+
+import io
 import os
-import urllib
+import sys
+import urllib.request, urllib.parse, urllib.error
 import time
 
 from datetime import datetime
-from urlparse import urlparse
-
+from urllib.parse import urlparse
 from flask import Request, Response
 from OpenSSL import SSL
 
@@ -38,7 +41,7 @@ def parse_request_line(line):
     items[0] = items[0].upper()
     if len(items) == 2:
         return tuple(items) + ('0.9',)
-    items[1] = urllib.unquote(items[1])
+    items[1] = urllib.parse.unquote(items[1])
     items[2] = items[2].split('/')[-1].strip()
     return tuple(items)
 
@@ -105,7 +108,7 @@ class (with appropriate request_handler established during
                 env.update({
                     'wsgi.version' : (1,0),
                     'wsgi.url_scheme' : 'http', # XXX incomplete
-                    'wsgi.input' : cStringIO.StringIO(''.join(body)),
+                    'wsgi.input' : io.BytesIO(b''.join(body)),
                     'wsgi.errors' : FileLikeErrorLogger(hlog),
                     'wsgi.multithread' : False,
                     'wsgi.multiprocess' : False,
@@ -148,9 +151,10 @@ def send_response(self, resp, version=(1,1)):
             resp.headers.set('Content-Length', str(size))
         else:
             sendfile = None
-
-        send("HTTP/%s %s %s\r\n" % (('%s.%s' % version), resp.status_code, resp.status))
-        send(str(resp.headers))
+        if 'Content-Length' not in resp.headers:
+            resp.headers['Content-Length'] = resp.calculate_content_length()
+        send(('HTTP/%i.%i %i %s\r\n' % (version + (resp.status_code, resp.status))).encode(resp.charset))
+        send(str(resp.headers).encode(resp.charset))
 
         if sendfile:
             send(open(sendfile, 'rb')) # diesel can stream fds
@@ -195,22 +199,27 @@ class HttpClient(TCPClient):
     def request(self, method, url, headers=None, body=None, timeout=None):
         '''Issues a `method` request to `path` on the
         connected server.  Sends along `headers`, and
-        body.
+        `body`.
 
-        Very low level--you must set "host" yourself,
-        for example.  It will set Content-Length,
-        however.
+        :param headers: Request's headers. If not provided, `Host` and
+        `Content-Length` will be automatically provided
         '''
         headers = headers or {}
-        url_info = urlparse(url)
-        fake_wsgi = dict(
-        (cgi_name(n), str(v).strip()) for n, v in headers.iteritems())
+        url_info = urlparse(url if isinstance(url, str) else url.decode())
+        fake_wsgi = dict((cgi_name(n), str(v).strip()) for n, v in headers.items())
+
+        if 'HTTP_HOST' not in fake_wsgi:
+            # HTTP host header omit the port if 80
+            # http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.23
+            if self.port == 80:
+                fake_wsgi['HTTP_HOST'] = self.addr
+            else:
+                fake_wsgi['HTTP_HOST'] = '%s:%i' % (self.addr, self.port)
 
         if body and 'CONTENT_LENGTH' not in fake_wsgi:
             # If the caller hasn't set their own Content-Length but submitted
             # a body, we auto-set the Content-Length header here.
-            fake_wsgi['CONTENT_LENGTH'] = str(len(body))
-
+            fake_wsgi['CONTENT_LENGTH'] = str(len(body) if body else 0)
         fake_wsgi.update({
             'REQUEST_METHOD' : method,
             'SCRIPT_NAME' : '',
@@ -218,7 +227,7 @@ def request(self, method, url, headers=None, body=None, timeout=None):
             'QUERY_STRING' : url_info[4],
             'wsgi.version' : (1,0),
             'wsgi.url_scheme' : 'http', # XXX incomplete
-            'wsgi.input' : cStringIO.StringIO(body or ''),
+            'wsgi.input' : io.BytesIO(body or b''),
             'wsgi.errors' : FileLikeErrorLogger(hlog),
             'wsgi.multithread' : False,
             'wsgi.multiprocess' : False,
@@ -228,14 +237,14 @@ def request(self, method, url, headers=None, body=None, timeout=None):
 
         timeout_handler = TimeoutHandler(timeout or 60)
 
-        url = str(req.path)
+        # Werkzeug desn't encode path but encodes query_string...
+        url = req.path
         if req.query_string:
-            url += '?' + str(req.query_string)
-
-        send('%s %s HTTP/1.1\r\n%s' % (req.method, url, str(req.headers)))
+            url += '?' + req.query_string.decode('latin-1')
+        send(('%s %s HTTP/1.1\r\n%s' % (req.method, url, str(req.headers))).encode(req.charset))
 
         if body:
-            send(body)
+            send(body.encode())
 
         h = HttpParser()
         body = []
@@ -253,7 +262,7 @@ def request(self, method, url, headers=None, body=None, timeout=None):
             data = val
 
         resp = Response(
-            response=''.join(body),
+            response=b''.join(body),
             status=h.get_status_code(),
             headers=h.get_headers(),
             )
diff --git a/diesel/protocols/http/pool.py b/diesel/protocols/http/pool.py
index fee8a86..07705a2 100644
--- a/diesel/protocols/http/pool.py
+++ b/diesel/protocols/http/pool.py
@@ -1,5 +1,5 @@
-import urllib
-import urlparse
+import urllib.request, urllib.parse, urllib.error
+import urllib.parse
 
 import diesel.protocols.http.core as http
 import diesel.util.pool as pool
@@ -19,11 +19,11 @@ class InvalidUrlScheme(Exception):
     pass
 
 def request(url, method='GET', timeout=60, body=None, headers=None):
-    if body and (not isinstance(body, basestring)):
-        body_bytes = urllib.urlencode(body)
+    if body and (not isinstance(body, str)):
+        body_bytes = urllib.parse.urlencode(body)
     else:
         body_bytes = body
-    req_url = urlparse.urlparse(url)
+    req_url = urllib.parse.urlparse(url)
     if not headers:
         headers = {}
     headers.update({
@@ -40,12 +40,12 @@ def request(url, method='GET', timeout=60, body=None, headers=None):
         req_path = req_url.path
     encoded_path = req_path.encode('utf-8')
     # Loop to retry if the connection was closed.
-    for i in xrange(POOL_SIZE):
+    for i in range(POOL_SIZE):
         try:
             with http_pool_for_url(req_url).connection as conn:
                 resp = conn.request(method, encoded_path, headers, timeout=timeout, body=body_bytes)
             break
-        except ClientConnectionClosed, e:
+        except ClientConnectionClosed as e:
             # try again with another pool connection
             continue
     else:
diff --git a/diesel/protocols/irc.py b/diesel/protocols/irc.py
index 1e4c101..af84b99 100644
--- a/diesel/protocols/irc.py
+++ b/diesel/protocols/irc.py
@@ -160,7 +160,7 @@ def _handle_return(self, r, chan):
             assert '\n' not in r
             assert '\0' not in r
             self.send_command('PRIVMSG', chan, r)
-        elif type(r) is unicode:
+        elif type(r) is str:
             self._handle_return(r.encode('utf-8'), chan)
         elif type(r) is tuple:
             chan, r = r
@@ -169,7 +169,7 @@ def _handle_return(self, r, chan):
             for i in r:
                 self._handle_return(i, chan)
         else:
-            print 'Hmm, unknown type returned from message handler:', type(r)
+            print('Hmm, unknown type returned from message handler:', type(r))
 
 class SSLIrcBot(IrcBot):
     def __init__(self, *args, **kw):
diff --git a/diesel/protocols/mongodb.py b/diesel/protocols/mongodb.py
index 37e07ab..2a34a0e 100644
--- a/diesel/protocols/mongodb.py
+++ b/diesel/protocols/mongodb.py
@@ -2,8 +2,9 @@
 """A mongodb client library for Diesel"""
 
 # needed to make diesel work with python 2.5
-from __future__ import with_statement
 
+
+from builtins import object
 import itertools
 import struct
 from collections import deque
@@ -14,14 +15,18 @@
 from diesel.transports.common import protocol, ConnectionClosed
 from diesel.transports.tcp import TCPClient
 
-_ZERO = "\x00\x00\x00\x00"
+
+_ZERO = b"\x00\x00\x00\x00"
 HEADER_SIZE = 16
 
+
 class MongoOperationalError(Exception): pass
 
+
 def _full_name(parent, child):
     return "%s.%s" % (parent, child)
 
+
 class TraversesCollections(object):
     def __init__(self, name, client):
         self.name = name
@@ -38,6 +43,7 @@ def __getitem__(self, name):
 class Db(TraversesCollections):
     pass
 
+
 class Collection(TraversesCollections):
     def find(self, spec=None, fields=None, skip=0, limit=0):
         return MongoCursor(self.name, self.client, spec, fields, skip, limit)
@@ -51,6 +57,7 @@ def insert(self, doc_or_docs, safe=True):
     def delete(self, spec, safe=True):
         return self.client.delete(self.name, spec, safe)
 
+
 class MongoClient(TCPClient):
     collection_class = None
 
@@ -60,7 +67,7 @@ def __init__(self, host='localhost', port=27017, **kw):
 
     @property
     def _msg_id(self):
-        return self._msg_id_counter.next()
+        return next(self._msg_id_counter)
 
     def _put_request_get_response(self, op, data):
         self._put_request(op, data)
@@ -74,7 +81,7 @@ def _put_request_get_response(self, op, data):
 
     def _put_request(self, op, data):
         req = struct.pack('<4i', HEADER_SIZE + len(data), self._msg_id, 0, op)
-        send("%s%s" % (req, data))
+        send(req + data)
 
     def _handle_response(self, cursor, resp):
         cid, start, numret, result = resp
@@ -158,8 +165,12 @@ def _command(self, dbname, command):
             return []
 
     def __getattr__(self, name):
+        return self[name]
+
+    def __getitem__(self, name):
         return Db(name, self)
 
+
 class Ops(object):
     ASCENDING = 1
     DESCENDING = -1
@@ -184,7 +195,7 @@ def query(col, spec, fields, skip, limit):
                 data.append(BSON.encode(fields))
             else:
                 data.append(BSON.encode(dict.fromkeys(fields, 1)))
-        return "".join(data)
+        return b"".join(data)
 
     @staticmethod
     def get_more(col, limit, id):
@@ -203,7 +214,7 @@ def update(col, spec, doc, upsert, multi):
             flags |= 1 << 1
         fmt = '<i%dsi' % len(colname)
         part = struct.pack(fmt, 0, colname, flags)
-        return "%s%s%s" % (part, BSON.encode(spec), BSON.encode(doc))
+        return part + BSON.encode(spec) + BSON.encode(doc)
 
     @staticmethod
     def insert(col, doc_or_docs):
@@ -212,21 +223,22 @@ def insert(col, doc_or_docs):
             doc_or_docs = [doc_or_docs]
         except AttributeError:
             pass
-        doc_data = "".join(BSON.encode(doc) for doc in doc_or_docs)
+        doc_data = b"".join(BSON.encode(doc) for doc in doc_or_docs)
         colname = _make_c_string(col)
-        return "%s%s%s" % (_ZERO, colname, doc_data)
+        return _ZERO + colname + doc_data
 
     @staticmethod
     def delete(col, spec):
         colname = _make_c_string(col)
-        return "%s%s%s%s" % (_ZERO, colname, _ZERO, BSON.encode(spec))
+        return _ZERO + colname + _ZERO + BSON.encode(spec)
+
 
 class MongoIter(object):
     def __init__(self, cursor):
         self.cursor = cursor
         self.cache = deque()
 
-    def next(self):
+    def __next__(self):
         try:
             return self.cache.popleft()
         except IndexError:
@@ -235,7 +247,8 @@ def next(self):
                 raise StopIteration()
             else:
                 self.cache.extend(more)
-                return self.next()
+                return next(self)
+        
 
 class MongoCursor(object):
     def __init__(self, col, client, spec, fields, skip, limit):
@@ -313,6 +326,7 @@ def __exit__(self, *args, **params):
         if self.id and not self.finished:
             raise RuntimeError("need to cleanup cursor!")
 
+
 class RawMongoClient(TCPClient):
     "A mongodb client that does the bare minimum to push bits over the wire."
 
@@ -328,6 +342,7 @@ def send(self, data, respond=False):
             body = receive(length - HEADER_SIZE)
             return header + body
 
+
 class MongoProxy(object):
     ClientClass = RawMongoClient
 
@@ -361,7 +376,7 @@ def __call__(self, service, addr):
 
     def handle_request(self, info, body):
         length, id, to, opcode = info
-        print "saw request with opcode", opcode
+        print("saw request with opcode", opcode)
         return None, info, body
 
     def handle_response(self, response):
diff --git a/diesel/protocols/nitro.py b/diesel/protocols/nitro.py
index be42ae3..73a64c8 100644
--- a/diesel/protocols/nitro.py
+++ b/diesel/protocols/nitro.py
@@ -176,7 +176,7 @@ def _handle_client_requests_and_responses(self, remote_client):
                     remote_client.async_frame = value.orig_frame
                     resp = self.handle_client_packet(value.data, remote_client.context)
                     if resp:
-                        if isinstance(resp, basestring):
+                        if isinstance(resp, str):
                             output = [resp]
                         else:
                             output = iter(resp)
diff --git a/diesel/protocols/pg.py b/diesel/protocols/pg.py
index 0a6f335..f032f24 100644
--- a/diesel/protocols/pg.py
+++ b/diesel/protocols/pg.py
@@ -171,7 +171,7 @@ def handle_row_description(self):
 
         names = []
         types = []
-        for x in xrange(n):
+        for x in range(n):
             name = until('\0')[:-1]
             names.append(name)
             taboid, fattnum, typoid, sz, typmod, fmt = \
@@ -191,7 +191,7 @@ def close(self):
     def handle_data(self):
         (size, n) = unpack("!ih", receive(6))
         values = PgDataRow()
-        for x in xrange(n):
+        for x in range(n):
             (l,) = unpack('!i', receive(4))
             if l == -1:
                 values.append(None)
@@ -233,7 +233,7 @@ def _query(self, q):
 
     @protocol
     def prepare(self, q, id):
-        print 'PREP'
+        print('PREP')
         q += '\0'
         sid = id + '\0'
         send('P' + pack('!i', 4 + len(q) + len(sid) + 2))
@@ -281,7 +281,7 @@ def _queue_query(self, q, args):
         if q in self.prepared:
             id = self.prepared[q]
         else:
-            id = 'q%d' % self.prepare_gen.next()
+            id = 'q%d' % next(self.prepare_gen)
             self.prepare(q, id)
             self.prepared[q] = id
             prep = True
@@ -379,7 +379,7 @@ def py_to_pg(o):
     return {
         bool : lambda d: 't' if d else 'f',
         str : str,
-        unicode : lambda a: a.encode('utf8'),
+        str : lambda a: a.encode('utf8'),
         int : str,
     }[t](o)
 
@@ -390,8 +390,8 @@ def oid_to_conv(id):
         21 : int,
         23 : int,
         26 : str,
-        1042 : lambda s: unicode(s, 'utf-8'),
-        1043 : lambda s: unicode(s, 'utf-8'),
+        1042 : lambda s: str(s, 'utf-8'),
+        1043 : lambda s: str(s, 'utf-8'),
     }[id]
 
 if __name__ == '__main__':
@@ -406,19 +406,19 @@ def f():
                 #    print row.name
                 #    print row.id
                 for row in client.query("SELECT * FROM typtest", buffer=500):
-                    print row
+                    print(row)
 
             client.execute("UPDATE companies set name = $1 where id < $2",
             "marky co", 5)
 
-        print '\n\n~~~done!~~~'
+        print('\n\n~~~done!~~~')
 
     def g():
         with PostgreSQLClient(database="test", user="test") as client:
-            for x in xrange(500):
+            for x in range(500):
                 r = client.query_one(
                 "select * from counters where id = $1 for update", 1)
-        print 'done'
+        print('done')
 
 
     from diesel import quickstart
diff --git a/diesel/protocols/redis.py b/diesel/protocols/redis.py
index 48014b3..4b4c84b 100644
--- a/diesel/protocols/redis.py
+++ b/diesel/protocols/redis.py
@@ -1,5 +1,6 @@
 import time
 import uuid
+from builtins import (bytes, chr, str, int, object)
 
 from contextlib import contextmanager
 
@@ -19,9 +20,45 @@ def flatten_arg_pairs(l):
 class RedisError(Exception): pass
 
 class RedisClient(TCPClient):
-    def __init__(self, host='localhost', port=REDIS_PORT, password=None, **kw):
-        self.password = password
+    def __init__(self, host='localhost', port=REDIS_PORT, password=None,
+                 encoding='utf-8', encoding_errors='strict',
+                 decode_responses=False, **kw):
+        """
+        :param host: host of the redis server
+        :param port: port of the redis server
+        :param password: password of the redis server
+        :param decode_responses: If true, the incomming socket bytes will be
+        decoded using the `encoding` charset
+        :param encoding: Encoding to use to encode/decode data given the
+        socket level communication must be done with bytes
+        :param encoding_errors: errors handling when encoding unicode
+        """
         super(RedisClient, self).__init__(host, port, **kw)
+        self.password = password
+        self.encoding = encoding
+        self.encoding_errors = encoding_errors
+        self.decode_responses = decode_responses
+
+    def _decode_response(self, resp):
+        """If requested, convert response into unicode"""
+        if isinstance(resp, bytes) and self.decode_responses:
+            return resp.decode(self.encoding)
+        else:
+            return resp
+
+    def _encode(self, value):
+        "Return a bytestring representation of the value"
+        if isinstance(value, bytes):
+            return value
+        elif isinstance(value, int):
+            value = str(value)
+        elif isinstance(value, float):
+            value = repr(value)
+        elif not isinstance(value, str):
+            value = str(value)
+        if isinstance(value, str):
+            value = value.encode(self.encoding, self.encoding_errors)
+        return value
 
     ##################################################
     ### GENERAL OPERATIONS
@@ -207,13 +244,13 @@ def setex(self, k, tm, v):
 
     @protocol
     def mset(self, d):
-        self._send('MSET', list=flatten_arg_pairs(d.iteritems()))
+        self._send('MSET', list=flatten_arg_pairs(iter(d.items())))
         resp = self._get_response()
         return resp
 
     @protocol
     def msetnx(self, d):
-        self._send('MSETNX', list=flatten_arg_pairs(d.iteritems()))
+        self._send('MSETNX', list=flatten_arg_pairs(iter(d.items())))
         resp = self._get_response()
         return resp
 
@@ -456,7 +493,7 @@ def srandmember(self, key):
 
     def __pair_with_scores(self, resp):
         return [(resp[x], float(resp[x+1]))
-                for x in xrange(0, len(resp), 2)]
+                for x in range(0, len(resp), 2)]
 
     @protocol
     def zadd(self, key, score, member):
@@ -576,7 +613,7 @@ def hget(self, key, field):
     def hmset(self, key, d):
         if not d:
             return True
-        args = [key] + flatten_arg_pairs(d.iteritems())
+        args = [key] + flatten_arg_pairs(iter(d.items()))
 
         self._send('HMSET', list=args)
         resp = self._get_response()
@@ -589,7 +626,7 @@ def hmget(self, key, l):
         args = [key] + l
         self._send('HMGET', list=args)
         resp = self._get_response()
-        return dict(zip(l, resp))
+        return dict(list(zip(l, resp)))
 
     @protocol
     def hincrby(self, key, field, amt):
@@ -631,7 +668,7 @@ def hvals(self, key):
     def hgetall(self, key):
         self._send('HGETALL', key)
         resp = self._get_response()
-        return dict(resp[x:x+2] for x in xrange(0, len(resp), 2))
+        return dict(resp[x:x+2] for x in range(0, len(resp), 2))
 
     @protocol
     def hsetnx(self, key, field, value):
@@ -716,15 +753,15 @@ def get_from_subscriptions(self, wake_sig=None):
 
         -- OR -- None, if wake_sig was fired
 
-        NOTE: The message will always be a string.  Handle this as you see fit.
+        NOTE: The message will always be of bytes type.  Handle this as you see fit.
         NOTE: subscribe/unsubscribe acks are ignored here
         '''
         while True:
             r = self._get_response(wake_sig)
             if r:
-                if r[0] == 'message':
+                if r[0] == b'message':
                     return [r[1]] + r[1:]
-                elif r[0] == 'pmessage':
+                elif r[0] == b'pmessage':
                     return r[1:]
             else:
                 return None
@@ -746,20 +783,20 @@ def send_raw_command(self, arguments):
         self._send(cmd, list=rest)
 
         line_one = until_eol()
-        if line_one[0] in ('+', '-', ':'):
+        c = chr(bytes(line_one)[0])
+        if c in ('+', '-', ':'):
             return line_one
-
-        if line_one[0] == '$':
+        if c == '$':
             amt = int(line_one[1:])
             if amt == -1:
                 return line_one
             return line_one + receive(amt) + until_eol()
-        if line_one[0] == '*':
+        if c == '*':
             nargs = int(line_one[1:])
             if nargs == -1:
                 return line_one
             out = line_one
-            for x in xrange(nargs):
+            for x in range(nargs):
                 head = until_eol()
                 out += head
                 out += receive(int(head[1:])) + until_eol()
@@ -767,11 +804,11 @@ def send_raw_command(self, arguments):
 
     def _send(self, cmd, *args, **kwargs):
         if 'list' in kwargs:
-            args = kwargs['list']
-        all = (cmd,) + tuple(str(s) for s in args)
-        send('*%s\r\n' % len(all))
+            args = tuple(kwargs['list'])
+        all = tuple(map(self._encode, (cmd,) + args))
+        send(('*%s\r\n' % len(all)).encode())
         for i in all:
-            send(('$%s\r\n' % len(i)) + i + '\r\n')
+            send(('$%s\r\n' % len(i)).encode() + i + b'\r\n')
 
     def _get_response(self, wake_sig=None):
         if wake_sig:
@@ -782,9 +819,9 @@ def _get_response(self, wake_sig=None):
         else:
             fl = until_eol().strip()
 
-        c = fl[0]
+        c = chr(bytes(fl)[0])
         if c == '+':
-            return fl[1:]
+            return self._decode_response(fl[1:])
         elif c == '$':
             l = int(fl[1:])
             if l == -1:
@@ -792,26 +829,27 @@ def _get_response(self, wake_sig=None):
             else:
                 resp = receive(l)
                 until_eol() # noop
-            return resp
+            return self._decode_response(resp)
         elif c == '*':
             count = int(fl[1:])
             resp = []
             if count == -1:
                 return None
-            for x in xrange(count):
+            for x in range(count):
                 hl = until_eol()
-                assert hl[0] in ['$', ':', '+']
-                if hl[0] == '$':
+                c_hl = chr(bytes(hl)[0])
+                assert c_hl in ['$', ':', '+'], c_hl
+                if c_hl == '$':
                     l = int(hl[1:])
                     if l == -1:
                         resp.append(None)
                     else:
-                        resp.append(receive(l))
+                        resp.append(self._decode_response(receive(l)))
                         until_eol() # noop
-                elif hl[0] == ':':
+                elif c_hl == ':':
                     resp.append(int(hl[1:]))
-                elif hl[0] == '+':
-                    resp.append(hl[1:].strip())
+                elif c_hl == '+':
+                    resp.append(self._decode_response(hl[1:].strip()))
             return resp
         elif c == ':':
             return int(fl[1:])
@@ -881,9 +919,10 @@ class RedisLock(object):
     def __init__(self, client, key, timeout=30):
         assert timeout >= 2, 'Timeout must be greater than 2 to guarantee the transaction'
         self.client = client
-        self.key = key
+        assert isinstance(key, (str, bytes))
+        self.key = key if isinstance(key, bytes) else key.encode()
         self.timeout = timeout
-        self.me = str(uuid.uuid4())
+        self.me = uuid.uuid4().bytes
 
     def __enter__(self):
         trans = self.client.transaction(watch=[self.key])
@@ -915,7 +954,22 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 #########################################
 ## Hub, an abstraction of sub behavior, etc
 class RedisSubHub(object):
-    def __init__(self, host='127.0.0.1', port=REDIS_PORT, password=None):
+    def __init__(self, host='127.0.0.1', port=REDIS_PORT, password=None,
+                 encoding='utf-8', encoding_errors='strict',
+                 decode_responses=False):
+        """
+        :param host: host of the redis server
+        :param port: port of the redis server
+        :param password: password of the redis server
+        :param decode_responses: If true, the incomming socket bytes will be
+        decoded using the `encoding` charset
+        :param encoding: Encoding to use to encode/decode data given the
+        socket level communication must be done with bytes
+        :param encoding_errors: errors handling when encoding unicode
+        """
+        self.encoding = encoding
+        self.encoding_errors = encoding_errors
+        self.decode_responses = decode_responses
         self.host = host
         self.port = port
         self.password= password
@@ -924,14 +978,31 @@ def __init__(self, host='127.0.0.1', port=REDIS_PORT, password=None):
         self.sub_rms = []
         self.subs = {}
 
+    def _encode(self, value):
+        "Return a bytestring representation of the value"
+        if isinstance(value, bytes):
+            return value
+        elif isinstance(value, int):
+            value = str(value)
+        elif isinstance(value, float):
+            value = repr(value)
+        elif not isinstance(value, str):
+            value = str(value)
+        if isinstance(value, str):
+            value = value.encode(self.encoding, self.encoding_errors)
+        return value
+
     def make_client(self):
-        client = RedisClient(self.host, self.port, self.password)
+        client = RedisClient(self.host, self.port, self.password,
+                             encoding=self.encoding,
+                             encoding_errors=self.encoding_errors,
+                             decode_responses=self.decode_responses)
         if self.password != None:
             client.auth()
         return  client
 
     def __isglob(self, glob):
-        return '*' in glob or '?' in glob or ('[' in glob and ']' and glob)
+        return b'*' in glob or b'?' in glob or (b'[' in glob and b']' and glob)
 
     def __call__(self):
         with self.make_client() as conn:
@@ -990,6 +1061,7 @@ def __call__(self):
     def subq(self, classes):
         if type(classes) not in (set, list, tuple):
             classes = [classes]
+        classes = [self._encode(classe) for classe in classes]
 
         q = Queue()
 
@@ -1009,6 +1081,7 @@ def subq(self, classes):
     def sub(self, classes):
         if type(classes) not in (set, list, tuple):
             classes = [classes]
+        classes = [self._encode(classe) for classe in classes]
 
         hb = self
         q = Queue()
diff --git a/diesel/protocols/riak.py b/diesel/protocols/riak.py
index fd36f15..7e10804 100644
--- a/diesel/protocols/riak.py
+++ b/diesel/protocols/riak.py
@@ -163,7 +163,7 @@ def _subrequest(self, inq, outq):
             else:
                 try:
                     res = self.get(key)
-                except Exception, e:
+                except Exception as e:
                     outq.put((key, False, e))
                 else:
                     outq.put((key, True, res))
@@ -175,10 +175,8 @@ def get_many(self, keys, concurrency_limit=100, no_failures=False):
         outq = Queue()
         for k in keys:
             inq.put(k)
-
-        for x in xrange(min(len(keys), concurrency_limit)):
+        for x in range(min(len(keys), concurrency_limit)):
             fork(self._subrequest, inq, outq)
-
         failure = False
         okay, err = [], []
         for k in keys:
@@ -359,7 +357,7 @@ def put(self, bucket, key, value, indexes=None, **params):
             key=key,
             content=content,
         )
-        for name, value in params.iteritems():
+        for name, value in params.items():
             setattr(request, name, value)
         self._send(request)
         response = self._receive()
@@ -386,7 +384,7 @@ def index(self, bucket, idx, key, end=None):
         response = self._receive()
         if response:
             result = _to_dict(response)
-            if result.has_key('keys'):
+            if 'keys' in result:
                 return result['keys']
         return None
 
@@ -428,7 +426,7 @@ def set_bucket_props(self, bucket, props):
         """
         request = riak_palm.RpbSetBucketReq(bucket=bucket)
         bucket_props = riak_palm.RpbBucketProps()
-        for name, value in props.iteritems():
+        for name, value in props.items():
             setattr(bucket_props, name, value)
         request.props = bucket_props
         self._send(request)
@@ -501,7 +499,7 @@ def __init__(self, x, y):
         self.y = y
 
 if __name__ == '__main__':
-    import cPickle
+    import pickle
 
     from diesel import quickstart, quickstop, sleep
 
@@ -558,10 +556,10 @@ def resolve_longest(t1, v1, t2, v2):
         # Custom Bucket.
         class PickleBucket(Bucket): # lol
             def loads(self, raw_value):
-                return cPickle.loads(raw_value)
+                return pickle.loads(raw_value)
 
             def dumps(self, rich_value):
-                return cPickle.dumps(rich_value)
+                return pickle.dumps(rich_value)
 
             def resolve(self, t1, v1, t2, v2):
                 # Returns the value with the smallest different between points.
@@ -609,11 +607,11 @@ def resolve(self, t1, v1, t2, v2):
         try:
             # Tell Riak to require 10000 nodes to write this before success.
             c.put('testing', 'error!', 'oh noes!', w=10000)
-        except RiakErrorResp, e:
+        except RiakErrorResp as e:
             assert e.errcode == 1, e.errcode
             assert e.errmsg == '{n_val_violation,3}', e.errmsg
             assert repr(e) == "RiakErrorResp: {n_val_violation,3}", repr(e)
-        except Exception, e:
+        except Exception as e:
             assert 0, "UNEXPECTED EXCEPTION: %r" % e
         else:
             assert 0, "DID NOT RAISE"
diff --git a/diesel/protocols/websockets.py b/diesel/protocols/websockets.py
index 32e5d8b..d5900e3 100644
--- a/diesel/protocols/websockets.py
+++ b/diesel/protocols/websockets.py
@@ -78,7 +78,7 @@ def do_upgrade(self, req):
 
             protocol = req.headers.get('Sec-WebSocket-Protocol', None)
             key = req.headers.get('Sec-WebSocket-Key')
-            accept = b64encode(hashlib.sha1(key + self.GUID).digest())
+            accept = b64encode(hashlib.sha1((key + self.GUID).encode()).digest())
             headers = {
                 'Upgrade' : 'websocket',
                 'Connection' : 'Upgrade',
@@ -192,7 +192,7 @@ def handle_rfc_6455_frames(self, inq, outq):
                     payload = array('B', receive(length))
                     if disconnecting:
                         continue
-                    for i in xrange(len(payload)):
+                    for i in range(len(payload)):
                         payload[i] ^= mask[i % 4]
 
                     try:
@@ -208,7 +208,7 @@ def handle_rfc_6455_frames(self, inq, outq):
                         break
                     disconnecting = True
                 else:
-                    payload = dumps(val)
+                    payload = dumps(val).encode()
 
                     b1 = 0x80 | (1 & 0x0f) # FIN + opcode
 
@@ -219,7 +219,6 @@ def handle_rfc_6455_frames(self, inq, outq):
                         header = pack('>BBH', b1, 126, payload_len)
                     elif payload_len >= 65536:
                         header = pack('>BBQ', b1, 127, payload_len)
-
                     send(header + payload)
 
     def handle_non_rfc_frames(self, inq, outq):
diff --git a/diesel/protocols/wsgi.py b/diesel/protocols/wsgi.py
index 1189b19..d9944e8 100644
--- a/diesel/protocols/wsgi.py
+++ b/diesel/protocols/wsgi.py
@@ -21,7 +21,7 @@ def __init__(self, wsgi_callable, port=80):
 
     def _start_response(self, env, status, response_headers, exc_info=None):
         if exc_info:
-            raise exc_info[0], exc_info[1], exc_info[2]
+            raise exc_info[0](exc_info[1]).with_traceback(exc_info[2])
         else:
             r = env['diesel.response']
             r.status = status
diff --git a/diesel/protocols/zeromq.py b/diesel/protocols/zeromq.py
index cb8221c..3e40a9b 100644
--- a/diesel/protocols/zeromq.py
+++ b/diesel/protocols/zeromq.py
@@ -47,7 +47,7 @@ def send(self, message, flags=0):
         while True:
             try:
                 self.socket.send(message, zmq.NOBLOCK | flags)
-            except zmq.ZMQError, e:
+            except zmq.ZMQError as e:
                 if e.errno == EAGAIN:
                     self.handle_transition(True) # force re-evaluation of EVENTS
                     self.write_gate.wait()
@@ -63,7 +63,7 @@ def recv(self, copy=True):
             self.read_gate.wait()
             try:
                 m = self.socket.recv(zmq.NOBLOCK, copy=copy)
-            except zmq.ZMQError, e:
+            except zmq.ZMQError as e:
                 if e.errno == EAGAIN:
                     self.handle_transition(True) # force re-evaluation of EVENTS
                     self.read_gate.wait()
@@ -89,7 +89,7 @@ def process_fire(self, dc):
     def ready_early(self):
         try:
             m = self.socket.recv(zmq.NOBLOCK, copy=False)
-        except zmq.ZMQError, e:
+        except zmq.ZMQError as e:
             if e.errno == EAGAIN:
                 return False
             else:
@@ -202,7 +202,7 @@ def _handle_client_requests_and_responses(self, remote_client):
                 elif evt == 'sleep':
                     break
                 if resp:
-                    if isinstance(resp, basestring):
+                    if isinstance(resp, str):
                         output = [resp]
                     else:
                         output = iter(resp)
diff --git a/diesel/security.py b/diesel/security.py
index 43e67d1..4526e49 100644
--- a/diesel/security.py
+++ b/diesel/security.py
@@ -10,7 +10,7 @@ def shake():
             hub.enable_write(sock)
         except SSL.WantX509LookupError:
             pass
-        except Exception, e:
+        except Exception as e:
             hub.unregister(sock)
             next(e)
         else:
diff --git a/diesel/transports/common.py b/diesel/transports/common.py
index 0614dda..f5c5cd8 100644
--- a/diesel/transports/common.py
+++ b/diesel/transports/common.py
@@ -1,21 +1,26 @@
 from diesel import runtime
 from diesel.logmod import log
 
+
 class ClientConnectionError(Exception):
     '''Raised if a client cannot connect.'''
 
+
 class ClientConnectionTimeout(Exception):
     '''Raised if the client connection timed out before succeeding.'''
 
+
 class ConnectionClosed(Exception):
     '''Raised if the client closes the connection.'''
     def __init__(self, msg, data=None):
         super(ConnectionClosed, self).__init__(msg)
         self.data = data
 
+
 class NoAssociatedSocket(Exception):
     '''Raised if there is no socket for the current Loop.'''
 
+
 class ClientConnectionClosed(Exception):
     '''Raised if the remote server (for a Client call) closes the connection.'''
     def __init__(self, msg, data=None, addr=None, port=None):
@@ -30,6 +35,7 @@ def __str__(self):
             s += ' (addr=%s, port=%s)' % (self.addr, self.port)
         return s
 
+
 class Client(object):
     def __init__(self, addr, port, ssl_ctx=None, timeout=None, source_ip=None):
         self.ssl_ctx = ssl_ctx
@@ -69,6 +75,7 @@ def _resolve(self, name):
     def on_connect(self):
         pass
 
+
 class Service(object):
     '''A network service listening on a certain port, with a protocol
     implemented by a passed connection handler.
@@ -112,6 +119,7 @@ def listening(self):
     def validate_handler(self, handler):
         pass
 
+
 class SocketContext(object):
     def __init__(self, sock):
         self.hub = runtime.current_app.hub
@@ -175,6 +183,7 @@ def shutdown(self, remote_closed=False):
     def on_fork_child(self, parent, child):
         pass
 
+
 class protocol(object):
     def __init__(self, f, inst=None):
         self.f = f
@@ -198,7 +207,6 @@ def __call__(self, *args, **kw):
                 r = self.f(self.client, *args, **kw)
             finally:
                 current_loop.connection_stack.pop()
-        except ConnectionClosed, e:
+        except ConnectionClosed as e:
             raise ClientConnectionClosed(str(e), addr=self.client.addr, port=self.client.port)
         return r
-
diff --git a/diesel/transports/tcp.py b/diesel/transports/tcp.py
index c12e38d..dc03e60 100644
--- a/diesel/transports/tcp.py
+++ b/diesel/transports/tcp.py
@@ -32,7 +32,7 @@ def _setup_socket(self, ip, timeout, source_ip=None):
 
         try:
             sock.connect(remote_addr)
-        except socket.error, e:
+        except socket.error as e:
             if e.args[0] == errno.EINPROGRESS:
                 _private_connect(
                         self, ip, sock, self.addr, self.port, timeout=timeout)
@@ -63,7 +63,7 @@ def bind_and_listen(self):
 
         try:
             sock.bind((self.iface, self.port))
-        except socket.error, e:
+        except socket.error as e:
             self.handle_cannot_bind(str(e))
 
         sock.listen(self.LISTEN_QUEUE_SIZE)
@@ -77,7 +77,7 @@ def listening(self):
     def accept_new_connection(self):
         try:
             sock, addr = self.sock.accept()
-        except socket.error, e:
+        except socket.error as e:
             code, s = e
             if code in (errno.EAGAIN, errno.EINTR):
                 return
@@ -142,7 +142,7 @@ def handle_write(self):
             else:
                 try:
                     bsent = self.sock.send(data)
-                except socket.error, e:
+                except socket.error as e:
                     code, s = e
                     if code in (errno.EAGAIN, errno.EINTR):
                         self.pipeline.backup(data)
@@ -177,7 +177,7 @@ def handle_read(self):
             return
         try:
             data = self.sock.recv(BUFSIZE)
-        except socket.error, e:
+        except socket.error as e:
             code, s = e
             if code in (errno.EAGAIN, errno.EINTR):
                 return
@@ -267,7 +267,7 @@ def read_callback():
         except socket.error:
             try:
                 d = sock.recv(1)
-            except socket.error, e:
+            except socket.error as e:
                 if e.errno == errno.ECONNREFUSED:
                     d = ''
                 else:
diff --git a/diesel/transports/udp.py b/diesel/transports/udp.py
index 795f51a..692f679 100644
--- a/diesel/transports/udp.py
+++ b/diesel/transports/udp.py
@@ -4,9 +4,7 @@
 import sys
 import traceback
 import types
-
 from OpenSSL import SSL
-
 from collections import deque
 
 from diesel import runtime
@@ -15,12 +13,15 @@
         Client, Service, SocketContext, ConnectionClosed,
 )
 
+
 # 65535 - 8 (header) - 20 (ipv4 header)
 DATAGRAM_SIZE_MAX = 65507
 
+
 class UDPDispatchLoop(Loop):
     pass
 
+
 class UDPClient(Client):
     def __init__(self, addr, port, source_ip=None):
         super(UDPClient, self).__init__(addr, port, source_ip = source_ip)
@@ -40,6 +41,7 @@ def _setup_socket(self, ip, timeout, source_ip=None):
     def _resolve(self, addr):
         return addr
 
+
 class UDPService(Service):
     """A `Service` that communicates with remote clients via UDP."""
 
@@ -75,7 +77,7 @@ def bind_and_listen(self):
 
         try:
             sock.bind((self.iface, self.port))
-        except socket.error, e:
+        except socket.error as e:
             self.handle_cannot_bind(str(e))
         self.iface, self.port = sock.getsockname()
 
@@ -91,15 +93,17 @@ def bind_and_listen(self):
     def register(self, app):
         pass
 
-class _Datagram(str):
+
+class _Datagram(bytes):
     def __new__(self, payload, addr):
-        inst = str.__new__(self, payload)
+        inst = bytes.__new__(self, payload)
         inst.addr = addr
         return inst
 
     def __repr__(self):
         return "Datagram(%r, %r)" % (str.__str__(self), self.addr)
 
+
 class UDPContext(SocketContext):
     def __init__(self, sock, ip=None, port=None):
         super(UDPContext, self).__init__(sock)
@@ -131,7 +135,7 @@ def handle_write(self):
             dgram = self.outgoing.popleft()
             try:
                 bsent = self.sock.sendto(dgram, dgram.addr)
-            except socket.error, e:
+            except socket.error as e:
                 code, s = e
                 if code in (errno.EAGAIN, errno.EINTR):
                     self.outgoing.appendleft(dgram)
@@ -161,21 +165,21 @@ def handle_read(self):
         try:
             data, addr = self.sock.recvfrom(DATAGRAM_SIZE_MAX)
             dgram = _Datagram(data, addr)
-        except socket.error, e:
+        except socket.error as e:
             code, s = e
             if code in (errno.EAGAIN, errno.EINTR):
                 return
-            dgram = _Datagram('', (None, None))
+            dgram = _Datagram(b'', (None, None))
         except (SSL.WantReadError, SSL.WantWriteError, SSL.WantX509LookupError):
             return
         except SSL.ZeroReturnError:
-            dgram = _Datagram('', (None, None))
+            dgram = _Datagram(b'', (None, None))
         except SSL.SysCallError:
-            dgram = _Datagram('', (None, None))
+            dgram = _Datagram(b'', (None, None))
         except:
             sys.stderr.write("Unknown Error on recv():\n%s"
             % traceback.format_exc())
-            dgram = _Datagram('', (None, None))
+            dgram = _Datagram(b'', (None, None))
 
         if not dgram:
             self.shutdown(True)
@@ -212,5 +216,6 @@ def on_fork_child(self, parent, child):
         child_context.remote_addr = self.remote_addr
         child.connection_stack.append(child_context)
 
+
 class BadUDPHandler(Exception):
     """Thrown when an invalid UDP handler is passed to a UDPService"""
diff --git a/diesel/util/debugtools.py b/diesel/util/debugtools.py
index 46c505b..894f953 100644
--- a/diesel/util/debugtools.py
+++ b/diesel/util/debugtools.py
@@ -29,8 +29,8 @@ def print_greenlet_stacks():
         stack = ''.join(traceback.format_stack(fr))
         stacks[stack] += 1
         loops[stack] = obj
-    for stack, count in sorted(stacks.iteritems(), key=itemgetter(1)):
+    for stack, count in sorted(iter(stacks.items()), key=itemgetter(1)):
         loop = loops[stack]
         loop_id = address_stripper.sub('', str(loop.loop_callable))
-        print '[%d] === %s ===' % (count, loop_id)
-        print stack
+        print('[%d] === %s ===' % (count, loop_id))
+        print(stack)
diff --git a/diesel/util/patches/__init__.py b/diesel/util/patches/__init__.py
index cccfef2..041b6f6 100644
--- a/diesel/util/patches/__init__.py
+++ b/diesel/util/patches/__init__.py
@@ -1,1 +1,1 @@
-from requests_lib import enable_requests
+from .requests_lib import enable_requests
diff --git a/diesel/util/patches/requests_lib.py b/diesel/util/patches/requests_lib.py
index 04aab6f..6b539de 100644
--- a/diesel/util/patches/requests_lib.py
+++ b/diesel/util/patches/requests_lib.py
@@ -3,7 +3,7 @@
 It's a hack.
 
 """
-import httplib
+import http.client
 
 import diesel
 from diesel.resolver import DNSResolutionError
@@ -109,14 +109,14 @@ def __iter__(self):
     def close(self):
         self._client.close()
 
-class HTTPConnection(httplib.HTTPConnection):
+class HTTPConnection(http.client.HTTPConnection):
     def connect(self):
         try:
             self.sock = SocketLike(self.host, self.port)
         except DNSResolutionError:
             raise requests.ConnectionError
 
-class HTTPSConnection(httplib.HTTPSConnection):
+class HTTPSConnection(http.client.HTTPSConnection):
     def connect(self):
         try:
             kw = {'ssl_ctx': SSL.Context(SSL.SSLv23_METHOD)}
diff --git a/diesel/util/pool.py b/diesel/util/pool.py
index 0475e3a..1f0052d 100644
--- a/diesel/util/pool.py
+++ b/diesel/util/pool.py
@@ -30,7 +30,7 @@ def __init__(self, init_callable, close_callable, pool_size=5, pool_max=None, po
         self.poll_max_timeout = poll_max_timeout
         if pool_max:
             self.remaining_conns = Queue()
-            for _ in xrange(pool_max):
+            for _ in range(pool_max):
                 self.remaining_conns.inp.append(None)
         else:
             self.remaining_conns = InfiniteQueue()
@@ -118,7 +118,7 @@ def __call__(self):
         self.running = 0
         try:
             while True:
-                for x in xrange(self.concurrency - self.running):
+                for x in range(self.concurrency - self.running):
                     self.running += 1
                     fork(self.handler_wrap)
 
@@ -127,14 +127,14 @@ def __call__(self):
                     self.trigger.clear()
 
                 try:
-                    n = self.generator()
+                    n = next(self.generator)
                 except StopIteration:
                     break
 
                 self.q.put(n)
                 sleep()
         finally:
-            for x in xrange(self.concurrency):
+            for x in range(self.concurrency):
                 self.q.put(ThreadPoolDie)
             if self.finalizer:
                 self.finished.wait()
diff --git a/diesel/util/process.py b/diesel/util/process.py
index ffbb329..413a598 100644
--- a/diesel/util/process.py
+++ b/diesel/util/process.py
@@ -90,7 +90,7 @@ def wrapper(pipe):
                 except (SystemExit, KeyboardInterrupt):
                     pipe.close()
                     break
-                except Exception, e:
+                except Exception as e:
                     e.original_traceback = traceback.format_exc()
                     pipe.send(e)
 
@@ -189,7 +189,7 @@ def pool(self):
         ProcessPool to your application.
 
         """
-        for i in xrange(self.concurrency):
+        for i in range(self.concurrency):
             proc = spawn(self.handler)
             self.available_procs.put(proc)
             self.all_procs.append(proc)
@@ -206,28 +206,28 @@ def sleep_and_return(secs):
 
     def main():
         def waiting(ident):
-            print ident, "waiting ..."
+            print(ident, "waiting ...")
             t = sleep_pool(4)
-            print ident, "woken up after", t
+            print(ident, "woken up after", t)
 
         diesel.fork(waiting, 'a')
         diesel.fork(waiting, 'b')
         diesel.fork(waiting, 'c')
-        for i in xrange(11):
-            print "busy!"
+        for i in range(11):
+            print("busy!")
             diesel.sleep(1)
         div = spawn(lambda x,y: x/y)
         try:
             div(1,0)
-        except ZeroDivisionError, e:
+        except ZeroDivisionError as e:
             diesel.log.error(e.original_traceback)
-        print '^^ That was an intentional exception.'
+        print('^^ That was an intentional exception.')
         term(div)
         psleep = spawn(sleep_and_return)
         diesel.fork(psleep, 0.5)
         diesel.fork(psleep, 0.5)
         diesel.sleep(1)
-        print '^^ That was an intentional exception.'
+        print('^^ That was an intentional exception.')
         diesel.quickstop()
 
     diesel.quickstart(sleep_pool.pool, main)
diff --git a/diesel/util/queue.py b/diesel/util/queue.py
index 4d144a4..3da70a2 100644
--- a/diesel/util/queue.py
+++ b/diesel/util/queue.py
@@ -40,7 +40,7 @@ def get(self, waiting=True, timeout=None):
     def __iter__(self):
         return self
 
-    def next(self):
+    def __next__(self):
         return self.get()
 
     @property
diff --git a/diesel/util/streams.py b/diesel/util/streams.py
index b21dfd0..bfc7d12 100644
--- a/diesel/util/streams.py
+++ b/diesel/util/streams.py
@@ -1,5 +1,4 @@
-import thread
-
+import _thread
 from diesel.core import fork_from_thread
 from diesel.util.queue import Queue
 
@@ -15,5 +14,5 @@ def consume_stream(stream, q):
 
 def create_line_input_stream(fileobj):
     q = Queue()
-    thread.start_new_thread(consume_stream, (fileobj, q))
+    _thread.start_new_thread(consume_stream, (fileobj, q))
     return q
diff --git a/diesel/web.py b/diesel/web.py
index 21fea2e..b2e0900 100644
--- a/diesel/web.py
+++ b/diesel/web.py
@@ -1,6 +1,4 @@
-'''Slight wrapper around flask to fit the diesel
-
-mold.
+'''Slight wrapper around flask to fit the diesel mold.
 '''
 from flask import * # we're essentially republishing
 from werkzeug.debug import tbtools
@@ -69,7 +67,7 @@ def handle_request(self, req):
         with self.request_context(req):
             try:
                 response = self.full_dispatch_request()
-            except Exception, e:
+            except Exception as e:
                 self.log_exception(e)
                 try:
                     response = self.make_response(self.handle_exception(e))
diff --git a/examples/chat-redis.py b/examples/chat-redis.py
index 4b10314..be69fca 100644
--- a/examples/chat-redis.py
+++ b/examples/chat-redis.py
@@ -1,6 +1,6 @@
 import time, cgi
 
-from diesel import Service, Application, sleep, first, Loop
+from diesel import Application, sleep, first, Loop
 
 from diesel.web import DieselFlask
 from diesel.protocols.websockets import WebSocketDisconnect
@@ -107,14 +107,17 @@ def pubsub_socket(req, inq, outq):
     with hub.subq('foo') as group:
         while True:
             q, v = first(waits=[inq, group])
-            if q == inq: # getting message from client
-                print "(inq) %s" % v
+            if isinstance(v, WebSocketDisconnect): # getting a disconnect signal
+                print("(disconnect)")
+                return
+            elif q == inq: # getting message from client
+                print("(inq) %s" % v)
                 cmd = v.get("cmd", "")
                 if cmd=="":
-                    print "published message to %i subscribers" % c.publish("foo", dumps({
+                    print("published message to %i subscribers" % c.publish("foo", dumps({
                     'nick' : cgi.escape(v['nick'].strip()),
                     'message' : cgi.escape(v['message'].strip()),
-                    }))
+                    })))
                 else:
                     outq.put(dict(message="test bot"))
             elif q == group: # getting message for broadcasting
@@ -122,14 +125,12 @@ def pubsub_socket(req, inq, outq):
                 try:
                     msg = loads(msg_str)
                     data = dict(message=msg['message'], nick=msg['nick'])
-                    print "(outq) %s" % data
+                    print("(outq) %s" % data)
                     outq.put(data)
                 except JSONDecodeError:
-                    print "error decoding message %s" % msg_str
-            elif isinstance(v, WebSocketDisconnect): # getting a disconnect signal
-                return
+                    print("error decoding message %s" % msg_str)
             else:
-                print "oops %s" % v
+                print("oops %s" % v)
 
 app.diesel_app.add_loop(Loop(hub))
 app.run()
diff --git a/examples/clocker.py b/examples/clocker.py
index d2eee4a..7b054b0 100644
--- a/examples/clocker.py
+++ b/examples/clocker.py
@@ -1,17 +1,27 @@
-from diesel import Loop, fork, Application, sleep
+from diesel import Loop, fork, Application, sleep, quickstop, wait
 from diesel.util.stats import CPUStats
+from diesel.util.queue import Queue
+
+WORKERS = 3
+WORKER_QUEUE = Queue()
 
 def not_always_busy_worker():
     with CPUStats() as stats:
-        for _ in xrange(12):
-            for i in xrange(10000000): # do some work to forward cpu seconds
+        for _ in range(12):
+            for i in range(10000000): # do some work to forward cpu seconds
                 pass
             sleep(0.1) # give up control
+    print("cpu seconds ",  stats.cpu_seconds)
+    WORKER_QUEUE.put(None)
 
-    print "cpu seconds ",  stats.cpu_seconds
+def teardown():
+    for _ in range(WORKERS):
+        wait(WORKER_QUEUE)
+    quickstop()
 
 def spawn_busy_workers():
-    for _ in xrange(0,3):
+    fork(teardown)
+    for _ in range(WORKERS):
         fork(not_always_busy_worker)
 
 a = Application()
diff --git a/examples/combined.py b/examples/combined.py
index c9d154f..895b606 100644
--- a/examples/combined.py
+++ b/examples/combined.py
@@ -1,30 +1,36 @@
 import time
-from diesel import send, quickstart, quickstop, until, log
+
+from diesel import send, quickstart, quickstop, until_eol, log
 from diesel.transports.common import protocol
 from diesel.transports.tcp import TCPService, TCPClient
 
+
 def handle_echo(service, remote_addr):
     while True:
-        message = until('\r\n')
-        send("you said: %s" % message)
+        message = until_eol()
+        send(b"you said: " + message)
+
 
 class EchoClient(TCPClient):
     @protocol
     def echo(self, message):
-        send(message + '\r\n')
-        back = until("\r\n")
+        send(message + b'\r\n')
+        back = until_eol()
         return back
 
+
 log = log.name('echo-system')
 
+
 def do_echos():
     client = EchoClient('localhost', 8000)
     t = time.time()
-    for x in xrange(5000):
-        msg = "hello, world #%s!" % x
+    for x in range(5000):
+        msg = ("hello, world #%s!" % x).encode()
         echo_result = client.echo(msg)
-        assert echo_result.strip() == "you said: %s" % msg
+        assert echo_result.strip() == b"you said: " + msg
     log.info('5000 loops in {0:.2f}s', time.time() - t)
     quickstop()
 
+
 quickstart(TCPService(handle_echo, port=8000), do_echos)
diff --git a/examples/combined_tls.py b/examples/combined_tls.py
index 32c6850..3d65fd9 100644
--- a/examples/combined_tls.py
+++ b/examples/combined_tls.py
@@ -1,35 +1,42 @@
 from OpenSSL import SSL
 import time
-from diesel import send, quickstart, quickstop, until, log
+
+from diesel import send, quickstart, quickstop, until_eol, log
 from diesel.transports.common import protocol
 from diesel.transports.tcp import TCPService, TCPClient
 
+
 server_ctx = SSL.Context(SSL.TLSv1_METHOD)
 server_ctx.use_privatekey_file('snakeoil-key.pem')
 server_ctx.use_certificate_file('snakeoil-cert.pem')
 
+
 def handle_echo(service, remote_addr):
     while True:
-        message = until('\r\n')
-        send("you said: %s" % message)
+        message = until_eol()
+        send(b"you said: " + message)
+
 
 class EchoClient(TCPClient):
     @protocol
     def echo(self, message):
-        send(message + '\r\n')
-        back = until("\r\n")
+        send(message.encode() + b'\r\n')
+        back = until_eol()
         return back
 
+
 log = log.name('echo-system')
 
+
 def do_echos():
     with EchoClient('localhost', 8000, ssl_ctx=SSL.Context(SSL.TLSv1_METHOD)) as client:
         t = time.time()
-        for x in xrange(5000):
+        for x in range(5000):
             msg = "hello, world #%s!" % x
             echo_result = client.echo(msg)
-            assert echo_result.strip() == "you said: %s" % msg
+            assert echo_result.strip() == ("you said: %s" % msg).encode()
         log.info('5000 loops in {0:.2f}s', time.time() - t)
     quickstop()
 
+
 quickstart(TCPService(handle_echo, port=8000, ssl_ctx=server_ctx), do_echos)
diff --git a/examples/consolechat.py b/examples/consolechat.py
index e3b4c93..411b50f 100644
--- a/examples/consolechat.py
+++ b/examples/consolechat.py
@@ -4,22 +4,49 @@
 telnet, type your name, hit enter, then chat.  Invite
 a friend to do the same.
 '''
+from __future__ import print_function
 import sys
+from diesel.transports.common import ConnectionClosed
 from diesel import (
     Application, TCPService, until_eol, fire, first, send, TCPClient, protocol,
     thread, fork, Loop,
 )
 from diesel.util.queue import Queue
 
+
+ENCODING='utf-8'
+
+
+if sys.version_info[0] == 2:
+    def get_input(prompt):
+        return raw_input(prompt).decode(sys.stdin.encoding)
+else:
+    def get_input(prompt):
+        return input(prompt)
+
+
+def encode(msg):
+    return msg.encode(ENCODING)
+
+
+def decode(buff):
+    return buff.decode(ENCODING)
+
+
 def chat_server(service, addr):
-    my_nick = until_eol().strip()
-    while True:
-        evt, data = first(until_eol=True, waits=['chat_message'])
-        if evt == 'until_eol':
-           fire('chat_message', (my_nick, data.strip()))
-        else:
-            nick, message = data
-            send("<%s> %s\r\n"  % (nick, message))
+    my_nick = b'unamed'
+    try:
+        my_nick = until_eol().strip()
+        while True:
+            evt, data = first(until_eol=True, waits=['chat_message'])
+            if evt == 'until_eol':
+               fire('chat_message', (my_nick, data.strip()))
+            else:
+                nick, message = data
+                send(b'<' + nick + b'> ' + message + b'\r\n')
+    except ConnectionClosed:
+        print('%s has closed connexion' % decode(my_nick))
+
 
 class ChatClient(TCPClient):
     def __init__(self, *args, **kw):
@@ -27,39 +54,44 @@ def __init__(self, *args, **kw):
         self.input = Queue()
 
     def read_chat_message(self, prompt):
-        msg = raw_input(prompt)
+        msg = get_input(prompt)
         return msg
 
     def input_handler(self):
-        nick = thread(self.read_chat_message, "nick: ").strip()
+        nick = encode(thread(self.read_chat_message, "nick: ").strip())
         self.nick = nick
         self.input.put(nick)
         while True:
-            msg = thread(self.read_chat_message, "").strip()
+            msg = encode(thread(self.read_chat_message, "").strip())
             self.input.put(msg)
 
     @protocol
     def chat(self):
         fork(self.input_handler)
         nick = self.input.get()
-        send("%s\r\n" % nick)
+        send(nick + b'\r\n')
         while True:
             evt, data = first(until_eol=True, waits=[self.input])
             if evt == "until_eol":
-                print data.strip()
+                print(decode(data.strip()))
             else:
-                send("%s\r\n" % data)
+                send(data + b'\r\n')
+
 
 def chat_client():
     with ChatClient('localhost', 8000) as c:
         c.chat()
 
+
 app = Application()
-if sys.argv[1] == "server":
-    app.add_service(TCPService(chat_server, 8000))
-elif sys.argv[1] == "client":
-    app.add_loop(Loop(chat_client))
+USAGE = "USAGE: python %s [server|client]" % sys.argv[0]
+if len(sys.argv) == 2:
+    if sys.argv[1] == "server":
+        app.add_service(TCPService(chat_server, 8000))
+    elif sys.argv[1] == "client":
+        app.add_loop(Loop(chat_client))
+    else:
+        raise SystemExit(USAGE)
 else:
-    print "USAGE: python %s [server|client]" % sys.argv[0]
-    raise SystemExit(1)
+    raise SystemExit(USAGE)
 app.run()
diff --git a/examples/crawler.py b/examples/crawler.py
index 2a73308..7cd0e66 100644
--- a/examples/crawler.py
+++ b/examples/crawler.py
@@ -4,8 +4,13 @@
 '''
 
 import sys, time, re, os
-from urlparse import urlparse, urljoin
+try:
+    from urllib.parse import urlparse, urljoin
+except ImportError:
+    from urlparse import urlparse, urljoin
 
+if len(sys.argv) != 3:
+    raise SystemExit('usage: %s url folder' % sys.argv[0])
 url, folder = sys.argv[1:]
 
 schema, host, path, _, _, _ = urlparse(url)
@@ -27,7 +32,7 @@
 heads = {'Host' : host}
 
 def get_links(s):
-    for mo in url_exp.finditer(s):
+    for mo in url_exp.finditer(s.decode('latin-1')):
         lpath = mo.group(2)
         if ':' not in lpath and '..' not in lpath:
             if lpath.startswith('/'):
@@ -51,7 +56,7 @@ def write_file(lpath, body):
     lpath = (lpath if not lpath.endswith('/') else (lpath + 'index.html')).lstrip('/')
     lpath = os.path.join(folder, lpath)
     ensure_dirs(lpath)
-    open(lpath, 'w').write(body)
+    open(lpath, 'wb').write(body)
 
 def follow_loop(lpath):
     log.info(" -> %s" % lpath)
@@ -81,7 +86,6 @@ def stop():
     quickstop()
 
 t = time.time()
-
-pool = ThreadPool(CONCURRENCY, follow_loop, req_loop().next, stop)
+pool = ThreadPool(CONCURRENCY, follow_loop, req_loop(), stop)
 
 quickstart(pool)
diff --git a/examples/dhttpd b/examples/dhttpd
index be88f49..e2b31e8 100755
--- a/examples/dhttpd
+++ b/examples/dhttpd
@@ -14,6 +14,7 @@ log = log.name("dhttpd")
 BASE = '.'
 PORT = 8080
 
+print(__file__)
 app = DieselFlask(__file__)
 
 @app.route("/<path:file>", methods=["GET"])
diff --git a/examples/dispatch.py b/examples/dispatch.py
index da82b09..2617dd0 100644
--- a/examples/dispatch.py
+++ b/examples/dispatch.py
@@ -13,11 +13,11 @@ def worker(x):
             r[x] += 1
 
 def maker():
-    for x in xrange(500000):
+    for x in range(500000):
         d.dispatch(x)
         if x % 10000 == 0:
             sleep()
             log.info("values: {0}", r)
     quickstop()
 
-quickstart(maker, *(partial(worker, x) for x in xrange(WORKERS)))
+quickstart(maker, *(partial(worker, x) for x in range(WORKERS)))
diff --git a/examples/echo.py b/examples/echo.py
index 2ddde51..111e580 100644
--- a/examples/echo.py
+++ b/examples/echo.py
@@ -1,14 +1,19 @@
 # vim:ts=4:sw=4:expandtab
 '''Simple echo server.
 '''
+from __future__ import print_function
 from diesel import Application, TCPService, until_eol, send
+from diesel.transports.common import ConnectionClosed
 
 def hi_server(service, addr):
-    while 1:
-        inp = until_eol()
-        if inp.strip() == "quit":
-            break
-        send("you said %s" % inp)
+    try:
+        while True:
+            inp = until_eol()
+            if inp.strip() == b"quit":
+                break
+            send(b"you said " + inp)
+    except ConnectionClosed:
+        print('client has closed connexion')
 
 app = Application()
 app.add_service(TCPService(hi_server, 8013))
diff --git a/examples/event.py b/examples/event.py
index b67f357..bd3b0de 100644
--- a/examples/event.py
+++ b/examples/event.py
@@ -5,17 +5,17 @@
 
 def racer():
     pistol.wait()
-    print "CHAAARGE!"
+    print("CHAAARGE!")
 
 def starter():
-    print "Ready..."
+    print("Ready...")
     sleep(1)
-    print "Set..."
+    print("Set...")
     sleep(1)
-    print " ~~~~~~~~~ BANG ~~~~~~~~~ "
+    print(" ~~~~~~~~~ BANG ~~~~~~~~~ ")
     pistol.set()
     sleep(1)
-    print " ~~~~~~~~~ RACE OVER ~~~~~~~~~ "
+    print(" ~~~~~~~~~ RACE OVER ~~~~~~~~~ ")
     quickstop()
 
-quickstart(starter, [racer for x in xrange(8)])
+quickstart(starter, [racer for x in range(8)])
diff --git a/examples/fanout.py b/examples/fanout.py
index 5c5b7d8..e9f457a 100644
--- a/examples/fanout.py
+++ b/examples/fanout.py
@@ -13,11 +13,11 @@ def listener(x):
     with f.sub() as q:
         while True:
             v = q.get()
-            print '%s <- %s' % (x, v)
+            print('%s <- %s' % (x, v))
             cd.tick()
 
 def teller():
-    for x in xrange(EVENTS):
+    for x in range(EVENTS):
         sleep(2)
         f.pub(x)
 
@@ -27,4 +27,4 @@ def killer():
 
 from functools import partial
 quickstart(killer, teller,
-        *[partial(listener, x) for x in xrange(LISTENERS)])
+        *[partial(listener, x) for x in range(LISTENERS)])
diff --git a/examples/forker.py b/examples/forker.py
index bba4463..04580b2 100644
--- a/examples/forker.py
+++ b/examples/forker.py
@@ -2,13 +2,13 @@
 
 def sleep_and_print(num):
     sleep(1)
-    print num
+    print(num)
     sleep(1)
     a.halt()
 
 
 def forker():
-    for x in xrange(5):
+    for x in range(5):
         fork(sleep_and_print, x)
 
 a = Application()
diff --git a/examples/http_pool.py b/examples/http_pool.py
index d8ef72b..e2f40b3 100644
--- a/examples/http_pool.py
+++ b/examples/http_pool.py
@@ -3,19 +3,20 @@
 
 from diesel import quickstart, quickstop
 from diesel.protocols.http.pool import request
+from functools import reduce
 
 def f():
     t1 = time()
-    print request("http://example.iana.org/missing"), 'is missing?'
+    print(request("http://www.example.org/missing"), 'is missing?')
     t2 = time()
-    print request("http://example.iana.org/missing"), 'is missing?'
+    print(request("http://www.example.org/missing"), 'is missing?')
     t3 = time()
-    print request("http://example.iana.org/missing"), 'is missing?'
+    print(request("http://www.example.org/missing"), 'is missing?')
     t4 = time()
-    print request("http://example.iana.org/"), 'is found?'
+    print(request("http://www.example.org/"), 'is found?')
     t5 = time()
 
-    print 'First request should (probably) have been longer (tcp handshake) than subsequent 3 requests:'
+    print('First request should (probably) have been longer (tcp handshake) than subsequent 3 requests:')
     reduce(lambda t1, t2: sys.stdout.write("%.4f\n" % (t2 - t1)) or t2, (t1, t2, t3, t4, t5))
 
     quickstop()
diff --git a/examples/http.py b/examples/http_simple.py
similarity index 100%
rename from examples/http.py
rename to examples/http_simple.py
diff --git a/examples/keep_alive.py b/examples/keep_alive.py
index a94227b..6ecd771 100644
--- a/examples/keep_alive.py
+++ b/examples/keep_alive.py
@@ -1,7 +1,7 @@
 from diesel import Application, Loop
 
 def restart():
-    print "I should restart"
+    print("I should restart")
     a = b
 
 a = Application()
diff --git a/examples/newwait.py b/examples/newwait.py
index 7b64ef7..428f3e6 100644
--- a/examples/newwait.py
+++ b/examples/newwait.py
@@ -19,9 +19,9 @@ def make_and_wait():
         q, v = first(waits=both)
         assert v is None
         if q == q1:
-            print 'q1'
+            print('q1')
         elif q == q2:
-            print 'q2'
+            print('q2')
         else:
             assert 0
 
diff --git a/examples/nitro.py b/examples/nitro.py
index 2637fb0..3e5f015 100644
--- a/examples/nitro.py
+++ b/examples/nitro.py
@@ -13,7 +13,7 @@ def server():
 
 def client():
     with DieselNitroSocket(connect=loc) as sock:
-        for x in xrange(100000):
+        for x in range(100000):
             sock.send(NitroFrame("Hello, dude!"))
             m = sock.recv()
             assert m.data == "you said: Hello, dude!"
diff --git a/examples/nitro_echo_service.py b/examples/nitro_echo_service.py
index 613d2b3..2b15f01 100644
--- a/examples/nitro_echo_service.py
+++ b/examples/nitro_echo_service.py
@@ -6,19 +6,19 @@
 import uuid
 
 NUM_CLIENTS = 300
-cids = range(NUM_CLIENTS)
+cids = list(range(NUM_CLIENTS))
 dead = 0
 
 def echo_client():
     global dead
     id = str(uuid.uuid4())
     s = DieselNitroSocket(connect='tcp://127.0.0.1:4321')
-    for i in xrange(50):
+    for i in range(50):
         s.send(NitroFrame('%s|m%d' % (id, i)))
         r = s.recv()
         assert r.data == 'm%d:%d' % (i, i + 1)
     dead += 1
-    print 'done!', dead
+    print('done!', dead)
 
 class EchoService(DieselNitroService):
     def handle_client_packet(self, packet, ctx):
@@ -30,8 +30,8 @@ def parse_message(self, raw):
         return raw.split('|')
 
     def cleanup_client(self, client):
-        print 'client timed out', client.identity
+        print('client timed out', client.identity)
 
 echo_svc = EchoService('tcp://*:4321')
-diesel.quickstart(echo_svc.run, *(echo_client for i in xrange(NUM_CLIENTS)))
+diesel.quickstart(echo_svc.run, *(echo_client for i in range(NUM_CLIENTS)))
 
diff --git a/examples/queue_fairness_and_speed.py b/examples/queue_fairness_and_speed.py
index b2f751c..f139269 100644
--- a/examples/queue_fairness_and_speed.py
+++ b/examples/queue_fairness_and_speed.py
@@ -19,24 +19,24 @@ def worker():
         if val == shutdown:
             break
         num_processed += 1
-    fmt_args = (diesel.core.current_loop, num_processed)
-    print "%s, worker done (processed %d items)" % fmt_args
+    fmt_args = (diesel.core._current_loop, num_processed)
+    print("%s, worker done (processed %d items)" % fmt_args)
     dones.put('done')
 
 def main():
     start = time.time()
 
-    for i in xrange(NUM_ITEMS):
+    for i in range(NUM_ITEMS):
         q.put('item %d' % i)
-    for i in xrange(NUM_WORKERS):
+    for i in range(NUM_WORKERS):
         q.put(shutdown)
 
-    for i in xrange(NUM_WORKERS):
+    for i in range(NUM_WORKERS):
         diesel.fork_child(worker)
-    for i in xrange(NUM_WORKERS):
+    for i in range(NUM_WORKERS):
         dones.get()
 
-    print 'all workers done in %.2f secs' % (time.time() - start)
+    print('all workers done in %.2f secs' % (time.time() - start))
     diesel.quickstop()
 
 if __name__ == '__main__':
diff --git a/examples/queue.py b/examples/queue_simple.py
similarity index 89%
rename from examples/queue.py
rename to examples/queue_simple.py
index 6f3d77a..74e6022 100644
--- a/examples/queue.py
+++ b/examples/queue_simple.py
@@ -9,7 +9,7 @@ def putter():
     log = glog.name("putter")
 
     log.info("putting 100000 things on queue")
-    for x in xrange(100000):
+    for x in range(100000):
         q.put(x)
         sleep()
 
@@ -32,4 +32,4 @@ def manage():
     cd.wait()
     quickstop()
 
-quickstart(manage, putter, [getter for x in xrange(4)])
+quickstart(manage, putter, [getter for x in range(4)])
diff --git a/examples/redis_lock.py b/examples/redis_lock.py
index 9332c87..dc06c91 100644
--- a/examples/redis_lock.py
+++ b/examples/redis_lock.py
@@ -1,12 +1,12 @@
+"""Implement the Redis INCR command using a lock. Obviously this is inefficient, but it's a good
+example of how to use the RedisLock class"""
+
 import random
 
 from diesel import fork, quickstop, quickstart, sleep
 from diesel.protocols.redis import RedisClient, RedisTransactionError, RedisLock, LockNotAcquired
 
 
-"""Implement the Redis INCR command using a lock. Obviously this is inefficient, but it's a good
-example of how to use the RedisLock class"""
-
 key = 'test-lock-key'
 incr_key = 'test-incr-key'
 counter = 0
@@ -17,7 +17,6 @@
 sleep_factor = 1
 
 
-
 def take_lock():
     global counter
     client = RedisClient('localhost', 6379)
@@ -30,13 +29,17 @@ def take_lock():
     except LockNotAcquired:
         pass
 
+
 def main():
     client = RedisClient('localhost', 6379)
     client.delete(key)
     client.set(incr_key, 0)
 
-    for _ in xrange(500):
+    print('Starting spawning concurent increments')
+    for i in range(1, 501):
         fork(take_lock)
+        if not i % 100:
+            print('Spawned %i/500 increment tentatives' % i)
         if random.random() > 0.1:
             sleep(random.random() / 10)
     sleep(2)
diff --git a/examples/redispub.py b/examples/redispub.py
index af7a6ed..888466c 100644
--- a/examples/redispub.py
+++ b/examples/redispub.py
@@ -2,35 +2,41 @@
 '''Simple RedisSubHub client example.
 '''
 
-from diesel import Application, Loop, sleep
+from diesel import Application, Loop, sleep, quickstop, thread
 from diesel.protocols.redis import RedisSubHub, RedisClient
+from diesel.util.event import Countdown
 import time, sys
 
+cd = Countdown(2)
+
 def send_loop():
     c = RedisClient()
     sleep(1)
-
-    print 'SEND S', time.time()
-
-    for x in xrange(500):
+    print('[sending] started', time.time())
+    for x in range(500):
         c.publish("foo", "bar")
-
-    print 'SEND E', time.time()
-
-hub = RedisSubHub()
-
-def recv_loop():
-    print 'RECV S', time.time()
-    with hub.sub('foo') as poll:
-        for x in xrange(500):
-            q, content = poll.fetch()
-    print 'RECV E', time.time()
-
-a = Application()
-a.add_loop(Loop(hub)) # start up the sub loop
-if 'send' in sys.argv:
+    print('[sending] finished', time.time())
+
+def recv_loop_gen(id):
+    def recv_loop():
+        print('[receiving %s] started' % id, time.time())
+        with hub.sub('foo') as poll:
+            for x in range(500):
+                q, content = poll.fetch()
+        print('[receiving %s] finished' % id, time.time())
+        cd.tick()
+    return recv_loop
+
+def killer():
+    cd.wait()
+    quickstop()
+
+if __name__ == '__main__':
+    hub = RedisSubHub()
+    a = Application()
+    a.add_loop(Loop(hub)) # start up the sub loop
+    a.add_loop(Loop(killer))
     a.add_loop(Loop(send_loop))
-if 'recv' in sys.argv:    
-    a.add_loop(Loop(recv_loop))
-    a.add_loop(Loop(recv_loop))
-a.run()
+    a.add_loop(Loop(recv_loop_gen(1)))
+    a.add_loop(Loop(recv_loop_gen(2)))
+    a.run()
diff --git a/examples/resolve_names.py b/examples/resolve_names.py
index 1a64ac6..197d8f8 100644
--- a/examples/resolve_names.py
+++ b/examples/resolve_names.py
@@ -2,15 +2,15 @@
 from diesel import Application, Loop, sleep
 
 def resolve_the_google():
-    print 'started resolution!'
+    print('started resolution!')
     g_ip = resolve_dns_name("www.google.com")
-    print "www.google.com's ip is %s" % g_ip
+    print("www.google.com's ip is %s" % g_ip)
     try:
         bad_host = "www.g8asdf21oogle.com"
-        print "now checking %s" % bad_host
+        print("now checking %s" % bad_host)
         resolve_dns_name(bad_host)
     except DNSResolutionError:
-        print "yep, it failed as expected"
+        print("yep, it failed as expected")
     else:
         raise RuntimeError("The bad host resolved.  That's unexpected.")
     g_ip = resolve_dns_name("www.google.com")
@@ -21,7 +21,7 @@ def resolve_the_google():
 
 def stuff():
     while True:
-        print "doing stuff!"
+        print("doing stuff!")
         sleep(0.01)
 
 a = Application()
diff --git a/examples/santa.py b/examples/santa.py
index 9e0df50..9e3114a 100644
--- a/examples/santa.py
+++ b/examples/santa.py
@@ -1,9 +1,12 @@
 import random
+
 from diesel import Application, Loop, fire, wait, sleep
 
+
 deer_group = []
 elf_group = []
 
+
 def santa():
     while True:
         deer_ready, elves_ready = len(deer_group) == 9, len(elf_group) == 3
@@ -13,6 +16,7 @@ def santa():
             work_with_group('elf', elf_group, 'meet in my study')
         sleep(random.random() * 1)
 
+
 def actor(name, type, group, task, max_group, max_sleep):
     def actor_event_loop():
         while True:
@@ -20,26 +24,29 @@ def actor_event_loop():
             if len(group) < max_group:
                 group.append(name)
                 wait('%s-group-started' % type)
-                print "%s %s" % (name, task)
+                print("%s %s" % (name, task))
                 wait('%s-group-done' % type)
     return actor_event_loop
 
+
 def work_with_group(name, group, message):
-    print "Ho! Ho! Ho! Let's", message
+    print("Ho! Ho! Ho! Let's", message)
     fire('%s-group-started' % name)
     sleep(random.random() * 3)
     excuse_group(name, group)
 
+
 def excuse_group(name, group):
     group[:] = []
     fire('%s-group-done' % name, True)
 
+
 def main():
     app = Application()
     app.add_loop(Loop(santa))
 
     elf_do = "meets in study"
-    for i in xrange(10):
+    for i in range(10):
         app.add_loop(Loop(actor("Elf %d" % i, 'elf', elf_group, elf_do, 3, 3)))
 
     deer_do = "delivers toys"
@@ -52,5 +59,6 @@ def main():
 
     app.run()
 
+
 if __name__ == '__main__':
     main()
diff --git a/examples/sleep_server.py b/examples/sleep_server.py
index 989ded0..f1c00fe 100644
--- a/examples/sleep_server.py
+++ b/examples/sleep_server.py
@@ -6,10 +6,10 @@
 def delay_echo_server(service, addr):
     inp = until_eol()
 
-    for x in xrange(4):
+    for x in range(4):
         sleep(2)
-        send(str(x) + '\r\n')
-    send("you said %s" % inp)
+        send(str(x).encode() + b'\r\n')
+    send(b"you said " + inp)
 
 app = Application()
 app.add_service(TCPService(delay_echo_server, 8013))
diff --git a/examples/stdin.py b/examples/stdin.py
index 8161ed1..a6506ab 100644
--- a/examples/stdin.py
+++ b/examples/stdin.py
@@ -2,14 +2,14 @@
 
 from diesel import quickstart, fork_from_thread
 from diesel.util.queue import Queue
-from thread import start_new_thread
+from _thread import start_new_thread
 
 q = Queue()
 
 def consume():
     while True:
         v = q.get()
-        print 'DIESEL GOT', v
+        print('DIESEL GOT', v)
 
 def put(line):
     q.put(line)
@@ -17,7 +17,7 @@ def put(line):
 def create():
     while True:
         line = sys.stdin.readline()
-        print 'iter!', line
+        print('iter!', line)
         fork_from_thread(put, line)
 
 start_new_thread(create, ())
diff --git a/examples/stdin_stream.py b/examples/stdin_stream.py
index f0967b8..8d6f2d0 100644
--- a/examples/stdin_stream.py
+++ b/examples/stdin_stream.py
@@ -1,3 +1,4 @@
+from __future__ import print_function
 import sys
 
 from diesel import quickstart
@@ -7,6 +8,6 @@ def consume():
     q = create_line_input_stream(sys.stdin)
     while True:
         v = q.get()
-        print 'DIESEL GOT', v
+        print('DIESEL GOT', v)
 
 quickstart(consume)
diff --git a/examples/synchronized.py b/examples/synchronized.py
index 28d80f2..f5f3d6f 100644
--- a/examples/synchronized.py
+++ b/examples/synchronized.py
@@ -1,6 +1,9 @@
+from __future__ import print_function
+import random
+
 from diesel import Application, Loop, sleep
 from diesel.util.lock import synchronized
-import random
+
 free = 0
 sync = 0
 
@@ -9,7 +12,7 @@ def free_loop():
     free += 1
     sleep(random.random())
     free -= 1
-    print 'FREE', free
+    print('FREE', free)
 
 def sync_loop():
     global sync
@@ -17,7 +20,7 @@ def sync_loop():
         sync += 1
         sleep(random.random())
         sync -= 1
-        print 'SYNC', sync
+        print('SYNC', sync)
 
 def manage():
     sleep(10)
@@ -25,7 +28,7 @@ def manage():
 
 a = Application()
 for l in (free_loop, sync_loop):
-    for x in xrange(10):
+    for x in range(10):
         a.add_loop(Loop(l))
 a.add_loop(Loop(manage))
 a.run()
diff --git a/examples/thread_pool.py b/examples/thread_pool.py
index b41c5fc..7d517a3 100644
--- a/examples/thread_pool.py
+++ b/examples/thread_pool.py
@@ -3,15 +3,15 @@
 import random
 
 def handle_it(i):
-    print 'S', i
+    print('S', i)
     sleep(random.random())
-    print 'E', i
+    print('E', i)
 
 def c():
-    for x in xrange(0, 20):
+    for x in range(0, 20):
         yield x
 
-make_it = c().next
+make_it = c()
 
 def stop_it():
     quickstop()
diff --git a/examples/thread.py b/examples/thread_simple.py
similarity index 100%
rename from examples/thread.py
rename to examples/thread_simple.py
diff --git a/examples/timer.py b/examples/timer.py
index d781e09..bd13dec 100644
--- a/examples/timer.py
+++ b/examples/timer.py
@@ -2,8 +2,8 @@
 import time
 
 def l():
-    for x in xrange(2):
-        print "hi"
+    for x in range(2):
+        print("hi")
         time.sleep(1)
         sleep(5)
     a.halt()
diff --git a/examples/timer_bench.py b/examples/timer_bench.py
index 09ecacf..802e5e6 100644
--- a/examples/timer_bench.py
+++ b/examples/timer_bench.py
@@ -25,15 +25,15 @@
 
 
 def producer(q):
-    for i in xrange(OPERATIONS):
-        diesel.sleep(0.5)
+    for i in range(OPERATIONS):
+        diesel.sleep(0.1)
         q.put(i)
 
 def consumer(q, done):
-    for i in xrange(OPERATIONS):
+    for i in range(OPERATIONS):
         evt, data = diesel.first(waits=[q], sleep=10000)
         if evt == "sleep":
-            print "sleep was triggered!"
+            print("sleep was triggered!")
             break
     done.tick()
 
@@ -45,7 +45,7 @@ def pair(done):
 def track_cpu_stats():
     pid = os.getpid()
     def append_stats():
-        rawstats = subprocess.Popen(['ps -p %d -f' % pid], shell=True, stdout=subprocess.PIPE).communicate()[0]
+        rawstats = subprocess.Popen(['ps -p %d -f' % pid], shell=True, stdout=subprocess.PIPE).communicate()[0].decode()
         header, data = rawstats.split('\n', 1)
         procstats = [d for d in data.split(' ') if d]
         cpustats.append(int(procstats[3]))
@@ -56,16 +56,19 @@ def append_stats():
 def main():
     diesel.fork(track_cpu_stats)
     actor_pairs = int(sys.argv[1])
+    print('Running %i actors, with %i operations each' % (actor_pairs, OPERATIONS))
     done = Countdown(actor_pairs)
-    for i in xrange(actor_pairs):
+    for i in range(actor_pairs):
         pair(done)
     start = time.time()
     done.wait()
-    print "done in %.2f secs" % (time.time() - start)
+    print("done in %.2f secs" % (time.time() - start))
     diesel.sleep(1)
+    print('cupstats:', cpustats)
     diesel.quickstop()
 
 if __name__ == '__main__':
+    if len(sys.argv) != 2:
+        raise SystemExit('usage: %s <actor_pairs>' % sys.argv[0])
     diesel.set_log_level(diesel.loglevels.ERROR)
     diesel.quickstart(main)
-    print cpustats
diff --git a/examples/udp_echo.py b/examples/udp_echo.py
index 3077801..ba0806d 100644
--- a/examples/udp_echo.py
+++ b/examples/udp_echo.py
@@ -1,7 +1,9 @@
 # vim:ts=4:sw=4:expandtab
 '''Simple udp echo server and client.
 '''
+from __future__ import print_function
 import sys
+from builtins import input
 from diesel import (
     UDPService, UDPClient, protocol, send, datagram, quickstart, receive,
 )
@@ -16,7 +18,7 @@ class EchoClient(UDPClient):
     """
     @protocol
     def say(self, msg):
-        send(msg)
+        send(msg.encode())
         return receive(datagram)
 
 def echo_server(service):
@@ -30,13 +32,13 @@ def echo_server(service):
     """
     while True:
         data = receive(datagram)
-        send("you said %s" % data)
+        send(b"you said " + data)
 
 def echo_client():
     client = EchoClient('localhost', 8013)
     while True:
-        msg = raw_input("> ")
-        print client.say(msg)
+        msg = input("> ")
+        print(client.say(msg).decode())
 
 if len(sys.argv) == 2:
     if 'client' in sys.argv[1]:
@@ -45,4 +47,4 @@ def echo_client():
     elif 'server' in sys.argv[1]:
         quickstart(UDPService(echo_server, 8013))
         raise SystemExit
-print 'usage: python %s (server|client)' % sys.argv[0]
+print('usage: python %s (server|client)' % sys.argv[0])
diff --git a/examples/web.py b/examples/web.py
index 38c2b4e..e8733ca 100644
--- a/examples/web.py
+++ b/examples/web.py
@@ -17,6 +17,6 @@ def err():
     def t():
         while True:
             diesel.sleep(1)
-            print "also looping.."
+            print("also looping..")
     app.diesel_app.add_loop(diesel.Loop(t))
     app.run(debug=True)
diff --git a/examples/zeromq_echo_service.py b/examples/zeromq_echo_service.py
index 7a6cc45..ed36510 100644
--- a/examples/zeromq_echo_service.py
+++ b/examples/zeromq_echo_service.py
@@ -6,7 +6,8 @@
 
 
 NUM_CLIENTS = 100
-cids = range(NUM_CLIENTS)
+cids = list(range(NUM_CLIENTS))
+
 
 class EchoClient(ZeroMQClient):
     def __init__(self, ident, *args, **kw):
@@ -15,10 +16,12 @@ def __init__(self, ident, *args, **kw):
 
     @protocol
     def echo(self):
-        for i in xrange(10):
-            diesel.send('msg:%d' % i)
+        for i in range(10):
+            msg = ('msg:%d' % i).encode()
+            diesel.send(msg)
             r = diesel.receive()
-            assert r == 'msg:%d' % i
+            assert r == msg
+
 
 def client_loop(i):
     def wrapper():
@@ -27,6 +30,7 @@ def wrapper():
             client.echo()
     return wrapper
 
+
 def handle_messages(service):
     count = 0
     while True:
@@ -34,7 +38,8 @@ def handle_messages(service):
         diesel.send(msg)
         count += 1
         if count % 10 == 0:
-            print "handled", count, "messages"
+            print("handled %i messages" % count)
+
 
 echo_svc = ZeroMQService(zmq.REP, handle_messages, 4321)
 diesel.quickstart(echo_svc, *(client_loop(i) for i in xrange(NUM_CLIENTS)))
diff --git a/examples/zeromq_first.py b/examples/zeromq_first.py
index 2eb3d07..41c0388 100644
--- a/examples/zeromq_first.py
+++ b/examples/zeromq_first.py
@@ -5,13 +5,13 @@
 def get_messages():
     outsock = DieselZMQSocket(zctx.socket(zmq.DEALER), bind="tcp://127.0.0.1:5000")
 
-    for x in xrange(1000):
+    for x in range(1000):
         t, m = first(sleep=1.0, waits=[outsock])
 
         if t == 'sleep':
-            print "sleep timeout!"
+            print("sleep timeout!")
         else:
-            print "zmq:", m
+            print("zmq:", m)
 
     quickstop()
 
diff --git a/examples/zeromq_receiver.py b/examples/zeromq_receiver.py
index 09d36c0..2b1bef5 100644
--- a/examples/zeromq_receiver.py
+++ b/examples/zeromq_receiver.py
@@ -6,19 +6,19 @@ def get_messages():
     outsock = DieselZMQSocket(zctx.socket(zmq.DEALER), bind="tcp://127.0.0.1:5000")
 
     t = time.time()
-    for x in xrange(500000):
+    for x in range(500000):
         msg = outsock.recv()
         assert msg == "yo dawg %s" % x
         if x % 1000 == 0:
             sleep()
 
     delt = time.time() - t
-    print "500000 messages in %ss (%.1f/s)" % (delt, 500000.0 / delt)
+    print("500000 messages in %ss (%.1f/s)" % (delt, 500000.0 / delt))
     quickstop()
 
 def tick():
     while True:
-        print "Other diesel stuff"
+        print("Other diesel stuff")
         sleep(1)
 
 quickstart(get_messages, tick)
diff --git a/examples/zeromq_sender.py b/examples/zeromq_sender.py
index ffce86d..5a4d9d0 100644
--- a/examples/zeromq_sender.py
+++ b/examples/zeromq_sender.py
@@ -5,14 +5,14 @@
 def send_message():
     outsock = DieselZMQSocket(zctx.socket(zmq.DEALER), connect="tcp://127.0.0.1:5000")
 
-    for x in xrange(500000):
+    for x in range(500000):
         outsock.send("yo dawg %s" % x)
         if x % 1000 == 0:
             sleep()
 
 def tick():
     while True:
-        print "Other diesel stuff"
+        print("Other diesel stuff")
         sleep(1)
 
 quickstart(send_message, tick)
diff --git a/setup.py b/setup.py
index 51fa5ee..72eaf09 100644
--- a/setup.py
+++ b/setup.py
@@ -59,10 +59,14 @@
     },
     install_requires=([
         "greenlet",
-        "twiggy",
         "pyopenssl",
         "flask",
         "http-parser >= 0.7.12",
-        "dnspython",
-    ] + additional_requires),
+        "dnspython" if sys.version_info[0] == 2 else "dnspython3",
+        "simplejson",
+        "future",
+        "twiggy",
+        "pymongo",
+        "pyzmq"
+    ])
     )
diff --git a/tox.ini b/tox.ini
index 70b35be..2bf645f 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,8 +1,8 @@
 [tox]
-envlist = py26, py27
+envlist = py26, py27, py33, py34
 
 [testenv]
 deps =
     nose
 commands =
-    dnosetests tests/unit/ tests/integration/
+    dnosetests tests/unit/ tests/integration/ tests/protocol


Test patch:
diff --git a/examples/child_test.py b/examples/child_test.py
index 23a2e66..2323b3d 100644
--- a/examples/child_test.py
+++ b/examples/child_test.py
@@ -3,23 +3,23 @@
 def end_the_app_when_my_parent_dies():
     try:
         while True:
-            print "child: weeee!"
+            print("child: weeee!")
             sleep(1)
     except ParentDiedException:
-        print "child: ack, woe is me, I'm an orphan.  goodbye cruel world"
+        print("child: ack, woe is me, I'm an orphan.  goodbye cruel world")
         quickstop()
 
 
 def parent():
-    print "parent: okay, parent here."
+    print("parent: okay, parent here.")
     sleep(1)
-    print "parent: I'm so excited, about to become a parent"
+    print("parent: I'm so excited, about to become a parent")
     sleep(1)
     fork_child(end_the_app_when_my_parent_dies)
     sleep(1)
-    print "parent: and, there he goes.  I'm so proud"
+    print("parent: and, there he goes.  I'm so proud")
     sleep(4)
-    print "parent: okay, I'm outta here"
+    print("parent: okay, I'm outta here")
 
 
 if __name__ == '__main__':
diff --git a/examples/zeromq_test.py b/examples/zeromq_test.py
index 799adcd..f412b81 100644
--- a/examples/zeromq_test.py
+++ b/examples/zeromq_test.py
@@ -5,11 +5,11 @@
 def handle_messages():
     insock = DieselZMQSocket(zctx.socket(zmq.DEALER), bind="inproc://foo")
 
-    for x in xrange(500000):
+    for x in range(500000):
         msg = insock.recv()
         assert msg == "yo dawg %s" % x
     delt = time.time() - t
-    print "500000 messages in %ss (%.1f/s)" % (delt, 500000.0 / delt)
+    print("500000 messages in %ss (%.1f/s)" % (delt, 500000.0 / delt))
     quickstop()
 
 def send_message():
@@ -17,14 +17,14 @@ def send_message():
     outsock = DieselZMQSocket(zctx.socket(zmq.DEALER), connect="inproc://foo")
     t = time.time()
 
-    for x in xrange(500000):
+    for x in range(500000):
         outsock.send("yo dawg %s" % x)
         if x % 1000 == 0:
             sleep()
 
 def tick():
     while True:
-        print "Other diesel stuff"
+        print("Other diesel stuff")
         sleep(1)
 
 quickstart(handle_messages, send_message, tick)
diff --git a/tests/integration/test_event_ordering.py b/tests/integration/test_event_ordering.py
index 1d9bbb6..22b5a56 100644
--- a/tests/integration/test_event_ordering.py
+++ b/tests/integration/test_event_ordering.py
@@ -79,7 +79,7 @@ def delayed_value(self):
             "This connection requires more items in its delay list for further requests."
         delay = self.delay.pop(0)
         if delay is not None:
-            print diesel.sleep(delay)
+            print(diesel.sleep(delay))
         if self.waiting_callback:
             self.waiting_callback('expected value %s\r\n' % self.conn_id)
 
diff --git a/tests/integration/test_fanout.py b/tests/integration/test_fanout.py
index 6244485..d43407c 100644
--- a/tests/integration/test_fanout.py
+++ b/tests/integration/test_fanout.py
@@ -10,25 +10,25 @@ def setup(self):
         self.done = Countdown(10)
         self.fan = Fanout()
         self.subscriber_data = {}
-        for x in xrange(10):
+        for x in range(10):
             diesel.fork(self.subscriber)
         diesel.sleep()
-        for i in xrange(10):
+        for i in range(10):
             self.fan.pub(i)
         self.done.wait()
 
     def subscriber(self):
         self.subscriber_data[uuid.uuid4()] = data = []
         with self.fan.sub() as q:
-            for i in xrange(10):
+            for i in range(10):
                 data.append(q.get())
         self.done.tick()
 
 class TestFanout(FanoutHarness):
     def test_all_subscribers_get_the_published_messages(self):
         assert len(self.subscriber_data) == 10
-        for values in self.subscriber_data.itervalues():
-            assert values == range(10), values
+        for values in self.subscriber_data.values():
+            assert values == list(range(10)), values
 
     def test_sub_is_removed_after_it_is_done(self):
         assert not self.fan.subs
diff --git a/tests/integration/test_fork.py b/tests/integration/test_fork.py
index d5de171..d3aa5cb 100644
--- a/tests/integration/test_fork.py
+++ b/tests/integration/test_fork.py
@@ -23,12 +23,12 @@ def test_fork_many():
     v = [0]
     COUNT = 10000
     def parent():
-        for x in xrange(COUNT):
+        for x in range(COUNT):
             fork(tottering_child, v)
 
     runtime.current_app.add_loop(Loop(parent))
 
-    for i in xrange(16):
+    for i in range(16):
         if v[0] == COUNT:
             break
         sleep(0.5) # cumulative is long enough in core 2-era
diff --git a/tests/integration/test_os_signals.py b/tests/integration/test_os_signals.py
index 62bdde3..53f5a92 100644
--- a/tests/integration/test_os_signals.py
+++ b/tests/integration/test_os_signals.py
@@ -38,7 +38,7 @@ def test_multiple_signal_waiters():
     def mwaiter():
         diesel.signal(signal.SIGUSR1)
         c.tick()
-    for i in xrange(N_WAITERS):
+    for i in range(N_WAITERS):
         diesel.fork(mwaiter)
     diesel.sleep()
     os.kill(os.getpid(), signal.SIGUSR1)
@@ -94,7 +94,7 @@ def test_signal_captured_by_Signal_instance():
 
 def test_Signal_instances_trigger_multiple_times():
     usr1 = Signal(signal.SIGUSR1)
-    for i in xrange(5):
+    for i in range(5):
         os.kill(os.getpid(), signal.SIGUSR1)
         evt, _ = diesel.first(sleep=0.1, waits=[usr1])
         assert evt is usr1, evt
diff --git a/tests/integration/test_queue.py b/tests/integration/test_queue.py
index ad78e57..10af2d3 100644
--- a/tests/integration/test_queue.py
+++ b/tests/integration/test_queue.py
@@ -35,7 +35,7 @@ def worker(myid):
                 self.results.append(v)
                 self.handled[myid] += 1
                 self.done.tick()
-        for i in xrange(W):
+        for i in range(W):
             diesel.fork(worker, i)
 
     def trigger(self):
@@ -44,22 +44,22 @@ def trigger(self):
             assert 0, "timed out"
 
     def test_results_are_ordered_as_expected(self):
-        assert self.results == range(N), self.results
+        assert self.results == list(range(N)), self.results
 
     def test_results_are_balanced(self):
-        for wid, count in self.handled.iteritems():
+        for wid, count in self.handled.items():
             assert count == N/W, count
 
 class TestConsumersOnFullQueue(QueueHarness):
     def populate(self):
-        for i in xrange(N):
+        for i in range(N):
             self.queue.put(i)
 
 class TestConsumersOnEmptyQueue(QueueHarness):
     def populate(self):
         def go():
             diesel.wait('ready')
-            for i in xrange(N):
+            for i in range(N):
                 self.queue.put(i)
         diesel.fork(go)
 
diff --git a/tests/integration/test_sleep.py b/tests/integration/test_sleep.py
index 60939e4..1a679a4 100644
--- a/tests/integration/test_sleep.py
+++ b/tests/integration/test_sleep.py
@@ -37,7 +37,7 @@ def test_sleep_zero():
 
     v = [0]
     def i():
-        for i in xrange(10000):
+        for i in range(10000):
             v[0] += 1
             sleep()
 
diff --git a/tests/integration/test_tcp_transport.py b/tests/integration/test_tcp_transport.py
index 2539083..f8746cd 100644
--- a/tests/integration/test_tcp_transport.py
+++ b/tests/integration/test_tcp_transport.py
@@ -51,9 +51,9 @@ class EchoClient(TCPClient):
 
     @protocol
     def echo(self, msg):
-        diesel.send(msg + '\r\n')
+        diesel.send((msg + '\r\n').encode())
         result = diesel.until_eol()
-        return result.strip()
+        return result.strip().decode()
 
     def on_connect(self):
         self.on_connect_called = True
diff --git a/tests/integration/test_udp_transport.py b/tests/integration/test_udp_transport.py
index 0a050e3..01feb3f 100644
--- a/tests/integration/test_udp_transport.py
+++ b/tests/integration/test_udp_transport.py
@@ -15,6 +15,7 @@
 echo_service = None
 stream_service = None
 
+
 def setup_module():
     global echo_service, stream_service
     echo_service = UDPService(echo_handler, RANDOM_PORT, LOCAL_HOST)
@@ -23,21 +24,25 @@ def setup_module():
             stream_dispatcher, RANDOM_PORT, LOCAL_HOST, streaming=True)
     runtime.current_app.add_service(stream_service)
 
+
 def test_echo_client_can_talk_to_echo_service():
     client = EchoClient(LOCAL_HOST, echo_service.port)
     assert client.echo('hello') == 'hello'
 
+
 def test_client_can_make_multiple_requests():
     client = EchoClient(LOCAL_HOST, echo_service.port)
     assert client.echo('hello') == 'hello'
     assert client.echo('world') == 'world'
 
+
 def test_client_knows_if_it_is_closed_or_not():
     client = EchoClient(LOCAL_HOST, echo_service.port)
     assert not client.is_closed
     client.close()
     assert client.is_closed
 
+
 def test_service_interleaves_client_requests():
     client1 = EchoClient(LOCAL_HOST, echo_service.port)
     client2 = EchoClient(LOCAL_HOST, echo_service.port)
@@ -46,10 +51,11 @@ def test_service_interleaves_client_requests():
     assert client1.echo('HELLO') == 'HELLO'
     assert client2.echo('WORLD') == 'WORLD'
 
+
 def test_streaming_to_multiple_clients():
-    a_msgs = ['a:%d' % i for i in xrange(5)]
-    b_msgs = ['b:%d' % i for i in xrange(5)]
-    c_msgs = ['c:%d' % i for i in xrange(5)]
+    a_msgs = [('a:%d' % i).encode() for i in range(5)]
+    b_msgs = [('b:%d' % i).encode() for i in range(5)]
+    c_msgs = [('c:%d' % i).encode() for i in range(5)]
     expected = set()
     expected.update(a_msgs)
     expected.update(b_msgs)
@@ -58,15 +64,16 @@ def test_streaming_to_multiple_clients():
     actual_order = []
     results = Queue()
     def client_loop(actor, queue):
+        # import pdb; pdb.set_trace()
         s = StreamClient(queue, LOCAL_HOST, stream_service.port)
         s.start_stream(actor)
     diesel.fork(client_loop, 'a', results)
     diesel.fork(client_loop, 'b', results)
     diesel.fork(client_loop, 'c', results)
-    for i in xrange(15):
+    for i in range(15):
         msg = results.get(timeout=1.0)
-        actual_order.append(str(msg))
-        expected.remove(str(msg))
+        actual_order.append(bytes(msg))
+        expected.remove(bytes(msg))
     assert actual_order != unexpected_order, unexpected_order
     assert len(expected) == 0
 
@@ -79,6 +86,7 @@ def test_raises_exception_for_bad_service_handler():
     else:
         assert 0, "expected BadUDPHandler"
 
+
 def test_allows_instance_method_as_handler():
     class Foo(object):
         def handle(self, service):
@@ -88,6 +96,7 @@ def handle(self, service):
     except BadUDPHandler:
         assert 0, "instance method with two arguments should be ok"
 
+
 def test_allows_function_with_optional_arg_as_handler():
     def handle(service, foo=None):
         pass
@@ -102,23 +111,25 @@ def echo_handler(service):
     while True:
         diesel.send(diesel.receive(datagram))
 
+
 class EchoClient(UDPClient):
     @protocol
     def echo(self, msg):
-        diesel.send(msg)
+        diesel.send(msg.encode())
         result = diesel.receive(datagram)
-        return result.strip()
+        return result.strip().decode()
+
 
 def stream_dispatcher(service):
     def stream_handler(request):
-        for i in xrange(5):
-            diesel.send("%s:%d" % (request, i))
+        for i in range(5):
+            diesel.send(request + b':' + str(i).encode())
             diesel.sleep(random.random())
-
     while True:
         request = diesel.receive(datagram)
         fork_child(stream_handler, request)
 
+
 class StreamClient(UDPClient):
     def __init__(self, queue, *args, **kw):
         super(StreamClient, self).__init__(*args, **kw)
@@ -126,6 +137,6 @@ def __init__(self, queue, *args, **kw):
 
     @protocol
     def start_stream(self, data):
-        diesel.send(data)
-        for i in xrange(5):
+        diesel.send(data.encode())
+        for i in range(5):
             self.queue.put(diesel.receive(datagram))
diff --git a/tests/integration/test_wait.py b/tests/integration/test_wait.py
index c5104a9..5926280 100644
--- a/tests/integration/test_wait.py
+++ b/tests/integration/test_wait.py
@@ -20,7 +20,7 @@ def test_wait_tokens_dont_accumulate_forever():
     done = []
     wait_ids = []
     expected_length = len(diesel.runtime.current_app.waits.waits)
-    for i in xrange(50):
+    for i in range(50):
         wait_id = uuid.uuid4().hex
         diesel.fork(waiting_green_thread, wait_id, done)
         diesel.sleep()
diff --git a/tests/protocol/test_http.py b/tests/protocol/test_http.py
new file mode 100644
index 0000000..4cf3704
--- /dev/null
+++ b/tests/protocol/test_http.py
@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+
+from diesel import runtime, fork, sleep, wait, first, fire, quickstop, TCPService
+from diesel.transports.common import ClientConnectionClosed
+from diesel.protocols.wsgi import WSGIRequestHandler
+from diesel.protocols.http import HttpServer, HttpClient, Response
+
+try:
+    from urllib.parse import urlsplit
+except ImportError:
+    from urlparse import urlsplit
+
+
+class TestHttpClient(object):
+    def test_simple(self):
+        with HttpClient('httpbin.org', 80) as client:
+            resp = client.request('GET', '/get', {})
+        while resp.status_code == 302:
+            redirect = resp.headers['location']
+            assert redirect
+            url = urlsplit(redirect)
+            assert url.netloc, url
+            with HttpClient(url.netloc, 80) as client:
+                resp = client.request('GET', '%s?%s' % (url.path, url.query), {})
+        assert resp.status_code == 200, resp
+
+
+APP = None
+class TestWSGI(object):
+
+    SERVER_STARTED = False
+    PORT = 8000
+
+    def setup(self):
+        if TestWSGI.SERVER_STARTED:
+            return
+        def app_proxy(environ, start_response):
+            if not APP:
+                raise RuntimeError('missing app !')
+            return APP(environ, start_response)
+        http_service = TCPService(HttpServer(WSGIRequestHandler(app_proxy, self.PORT)), self.PORT, '')
+        runtime.current_app.add_service(http_service)
+        TestWSGI.SERVER_STARTED = True
+
+    def register_app(self, app):
+        global APP
+        APP = app
+
+    def test_simple(self):
+        def simple_app(environ, start_response):
+            """Simplest possible application object"""
+            status = '200 OK'
+            response_headers = [('Content-type','text/plain')]
+            start_response(status, response_headers)
+            return ["Hello World!"]
+        self.register_app(simple_app)
+        def client_app():
+            with HttpClient('localhost', 8000) as client:
+                try:
+                    resp = client.request('GET', '/', {})
+                except ClientConnectionClosed as e:
+                    fire('error', str(e))
+            fire('success', resp.data)
+        fork(client_app)
+        event, msg = first(waits=['success', 'error'])
+        assert event == 'success', msg
+        assert msg == b'Hello World!', msg
+
+    def test_unicode(self):
+        def unicode_app(environ, start_response):
+            status = '200 OK'
+            response_headers = [('Content-type','text/plain')]
+            start_response(status, response_headers)
+            return [u"你好!".encode('utf-8')]
+        self.register_app(unicode_app)
+        def client_app():
+            with HttpClient('localhost', 8000) as client:
+                try:
+                    resp = client.request('GET', '/', {})
+                except ClientConnectionClosed as e:
+                    fire('error', str(e))
+            fire('success', resp.data)
+        fork(client_app)
+        event, msg = first(waits=['success', 'error'])
+        assert event == 'success', msg
+        assert msg == u"你好!".encode('utf-8'), msg
+
+if __name__ == '__main__':
+    TestWSGI().test_simple()
diff --git a/tests/protocol/test_mongodb.py b/tests/protocol/test_mongodb.py
index 664d941..740463d 100644
--- a/tests/protocol/test_mongodb.py
+++ b/tests/protocol/test_mongodb.py
@@ -29,11 +29,11 @@ def test_insert_many(self):
         d1 = {'one' : 'two'}
         d2 = {'three' : 'four'}
         inp = [d1, d2]
-        inp.sort()
+        # inp.sort()
         assert self.db.imult.insert(inp)['err'] == None
         all = self.db.imult.find().all()
-        map(self.filt, all)
-        all.sort()
+        list(map(self.filt, all))
+        # all.sort()
         assert all == inp
 
     # UPDATE
@@ -141,7 +141,7 @@ def test_delete_all(self):
     def test_query_basic(self):
         d = {'one' : 'two'}
         assert self.db.onerec.insert(d)['err'] == None
-        assert map(self.filt, self.db.onerec.find().all()) == [d]
+        assert list(map(self.filt, self.db.onerec.find().all())) == [d]
         assert self.filt(self.db.onerec.find().one()) == d
 
         x = 0
@@ -166,21 +166,21 @@ def test_query_miss(self):
     def test_query_subfields(self):
         d = {'one' : 'two', 'three' : 'four'}
         assert self.db.q2.insert(d)['err'] == None
-        assert map(self.filt,
-                self.db.q2.find({'one' : 'two'}, ['three']).all()) \
+        assert list(map(self.filt,
+                self.db.q2.find({'one' : 'two'}, ['three']).all())) \
                 == [{'three' : 'four'}]
 
     def test_deterministic_order_when_static(self):
-        for x in xrange(500):
+        for x in range(500):
             self.db.q3.insert({'x' : x})
 
         snap = self.db.q3.find().all()
 
-        for x in xrange(100):
+        for x in range(100):
             assert snap == self.db.q3.find().all()
 
     def test_skip(self):
-        for x in xrange(500):
+        for x in range(500):
             self.db.q4.insert({'x' : x})
 
         snap = self.db.q4.find().all()
@@ -190,7 +190,7 @@ def test_skip(self):
         assert snap[250:] == with_skip
 
     def test_limit(self):
-        for x in xrange(500):
+        for x in range(500):
             self.db.q5.insert({'x' : x})
 
         snap = self.db.q5.find().all()
@@ -200,7 +200,7 @@ def test_limit(self):
         assert snap[:150] == with_skip
 
     def test_skip_limit(self):
-        for x in xrange(500):
+        for x in range(500):
             self.db.q6.insert({'x' : x})
 
         snap = self.db.q6.find().all()
@@ -211,7 +211,7 @@ def test_skip_limit(self):
 
     # CURSOR PROPERTIES
     def test_cursor_count(self):
-        for x in xrange(500):
+        for x in range(500):
             self.db.c1.insert({'x' : x})
 
         c = self.db.c1.find()
@@ -223,13 +223,12 @@ def test_cursor_count(self):
         assert len(c.all()) == 150
 
     def test_cursor_sort(self):
-        for x in xrange(500):
+        for x in range(500):
             self.db.c2.insert({'x' : x})
 
         snap = self.db.c2.find().sort('x', 1).all()
 
-        print snap
-        assert map(lambda d: d['x'], snap) == range(500)
+        assert [d['x'] for d in snap] == list(range(500))
 
         snap = self.db.c2.find().sort('x', -1).all()
-        assert map(lambda d: d['x'], snap) == range(499, -1, -1)
+        assert [d['x'] for d in snap] == list(range(499, -1, -1))
diff --git a/tests/protocol/test_redis.py b/tests/protocol/test_redis.py
index 405702b..438fe66 100644
--- a/tests/protocol/test_redis.py
+++ b/tests/protocol/test_redis.py
@@ -1,9 +1,12 @@
+# -*- coding: utf-8 -*-
+
 import diesel
 from diesel.protocols.redis import *
+from diesel.util.queue import QueueEmpty
 
 class RedisHarness(object):
     def setup(self):
-        self.client = RedisClient()
+        self.client = RedisClient(encoding='utf-8', decode_responses=True)
         self.client.select(11)
         self.client.flushdb()
 
@@ -18,7 +21,7 @@ def test_basic(self):
         r.delete('foo3')
         assert not r.exists('foo3')
 
-        for x in xrange(5000):
+        for x in range(5000):
             r.set('foo', 'bar')
 
         assert r.get('foo') == 'bar'
@@ -256,3 +259,79 @@ def test_transactions(self):
             assert transaction.aborted
         else:
             assert 0, "DID NOT RAISE"
+
+    def test_unicode(self):
+        r = self.client
+        assert r.get(u'明天') == None
+        r.set(u'明天', u'不好')
+        assert r.exists(u'明天')
+        assert r.get(u'明天') == r.get(u'明天'.encode('utf-8')) == u'不好' 
+        r.delete(u'明天')
+        assert not r.exists(u'明天')
+        r.set(u'明天', u'不好')
+        r.set(u'今天', u'很好')
+        assert r.keys(u'*天') == set([u'明天', u'今天'])
+        assert r.keys(u'不*') == set()
+
+
+class TestRedisNoDecodeResponses(object):
+    """RedisClient is configured to return raw bytes instead of decoded unicode"""
+    def setup(self):
+        self.client = RedisClient()
+        self.client.select(11)
+        self.client.flushdb()
+
+    def test_unicode(self):
+        encode = lambda x: x.encode('utf-8')
+        r = self.client
+        assert r.get(encode(u'明天')) == None
+        r.set(encode(u'明天'), encode(u'不好'))
+        assert r.exists(encode(u'明天'))
+        assert r.get(encode(u'明天')) == encode(u'不好')
+        r.delete(encode(u'明天'))
+        assert not r.exists(encode(u'明天'))
+        r.set(encode(u'明天'), encode(u'不好'))
+        r.set(encode(u'今天'), encode(u'很好'))
+        assert r.keys(encode(u'*天')) == set([encode(u'明天'), encode(u'今天')])
+        assert r.keys(encode(u'不*')) == set()
+
+class TestRedisSubHub(object):
+    def setup(self):
+        self.hub = RedisSubHub()
+        self.client = self.hub.make_client()
+        diesel.runtime.current_app.add_loop(diesel.Loop(self.hub))
+        self.client.select(11)
+        self.client.flushdb()
+
+    def test_subq(self):
+        messages = [b'1rst message', b'2nd message', b'3rd message']
+        queue = b'test_queue'
+        with self.hub.subq(queue) as group:
+            try:
+                group.get(waiting=False)
+            except QueueEmpty:
+                pass
+            else:
+                assert False, 'group queue should be empty'
+            def send_messages():
+                for message in messages:
+                    diesel.sleep(0.1)
+                    self.client.publish(queue, message)
+            diesel.fork(send_messages)
+            for expected_message in messages:
+                val = group.get(timeout=1)
+                assert val == (queue, expected_message), val
+
+    def test_sub(self):
+        event_listen = b'test_event_1'
+        event_ignore = b'test_event_2'
+        def send_event():
+            diesel.sleep(0.1)
+            self.client.publish(event_ignore, 'ignore me')
+            diesel.sleep(0.1)
+            self.client.publish(event_listen, 'event fired !')
+        diesel.fork(send_event)
+        with self.hub.sub(event_listen) as sub:
+            qn, msg = sub.fetch(1)
+            assert qn == event_listen, qn
+            assert msg == b'event fired !', msg
diff --git a/tests/unit/test_buffer.py b/tests/unit/test_buffer.py
index 1e16e24..a717748 100644
--- a/tests/unit/test_buffer.py
+++ b/tests/unit/test_buffer.py
@@ -2,42 +2,42 @@
 
 def test_feed():
     b = Buffer()
-    assert b.feed("rock") == None
-    assert b.feed("rockandroll") == None
+    assert b.feed(b"rock") == None
+    assert b.feed(b"rockandroll") == None
 
 def test_read_bytes():
     b = Buffer()
     b.set_term(10)
-    assert b.feed("rock") == None
-    assert b.feed("rockandroll") == "rockrockan"
+    assert b.feed(b"rock") == None
+    assert b.feed(b"rockandroll") == b"rockrockan"
     # buffer left.. droll
     assert b.check() == None
     b.set_term(10)
     assert b.check() == None
-    assert b.feed("r") == None
-    assert b.feed("r") == None
-    assert b.feed("r") == None
-    assert b.feed("r") == None
-    assert b.feed("r") == "drollrrrrr"
-    assert b.feed("x" * 10000) == None # no term (re-)established
+    assert b.feed(b"r") == None
+    assert b.feed(b"r") == None
+    assert b.feed(b"r") == None
+    assert b.feed(b"r") == None
+    assert b.feed(b"r") == b"drollrrrrr"
+    assert b.feed(b"x" * 10000) == None # no term (re-)established
 
 def test_read_sentinel():
     b = Buffer()
-    assert b.feed("rock and") == None
-    b.set_term("\r\n")
-    assert b.feed(" roll\r") == None
-    assert b.feed("\nrock ") == "rock and roll\r\n"
-    assert b.feed("and roll 2\r\n") == None
-    b.set_term("\r\n")
-    assert b.check() == "rock and roll 2\r\n"
+    assert b.feed(b"rock and") == None
+    b.set_term(b"\r\n")
+    assert b.feed(b" roll\r") == None
+    assert b.feed(b"\nrock ") == b"rock and roll\r\n"
+    assert b.feed(b"and roll 2\r\n") == None
+    b.set_term(b"\r\n")
+    assert b.check() == b"rock and roll 2\r\n"
 
 def test_read_hybrid():
     b = Buffer()
-    assert b.feed("rock and") == None
-    b.set_term("\r\n")
-    assert b.feed(" roll\r") == None
-    assert b.feed("\n012345678") == "rock and roll\r\n"
+    assert b.feed(b"rock and") == None
+    b.set_term(b"\r\n")
+    assert b.feed(b" roll\r") == None
+    assert b.feed(b"\n012345678") == b"rock and roll\r\n"
     b.set_term(16)
     assert b.check() == None
-    assert b.feed("9abcdefgh") == "0123456789abcdef"
+    assert b.feed(b"9abcdefgh") == b"0123456789abcdef"
 
diff --git a/tests/unit/test_pipeline.py b/tests/unit/test_pipeline.py
index 9567c2a..cbae7a6 100644
--- a/tests/unit/test_pipeline.py
+++ b/tests/unit/test_pipeline.py
@@ -1,7 +1,7 @@
 #12345678
 # That comment above matters (used in the test!)
 from diesel.pipeline import Pipeline, PipelineClosed, PipelineCloseRequest
-from cStringIO import StringIO
+from io import BytesIO
 
 FILE = __file__
 if FILE.endswith('.pyc') or FILE.endswith('.pyo'):
@@ -9,17 +9,17 @@
 
 def test_add_string():
     p = Pipeline()
-    assert (p.add("foo") == None)
+    assert (p.add(b"foo") == None)
     assert (not p.empty)
 
 def test_add_file():
     p = Pipeline()
-    assert (p.add(open(FILE)) == None)
+    assert (p.add(open(FILE, 'rb')) == None)
     assert (not p.empty)
 
 def test_add_filelike():
     p = Pipeline()
-    sio = StringIO()
+    sio = BytesIO()
     assert (p.add(sio) == None)
     assert (not p.empty)
 
@@ -36,69 +36,69 @@ class Whatever(object): pass
 
 def test_read_empty():
     p = Pipeline()
-    assert (p.read(500) == '')
+    assert (p.read(500) == b'')
 
 def test_read_string():
     p = Pipeline()
-    p.add("foo")
-    assert (p.read(3) == "foo")
+    p.add(b"foo")
+    assert (p.read(3) == b"foo")
     assert (p.empty)
 
 def test_read_file():
     p = Pipeline()
-    p.add(open(FILE))
-    assert (p.read(5) == "#1234")
+    p.add(open(FILE, 'rb'))
+    assert (p.read(5) == b"#1234")
 
 def test_read_filelike():
     p = Pipeline()
-    p.add(StringIO('abcdef'))
-    assert (p.read(5) == 'abcde')
+    p.add(BytesIO(b'abcdef'))
+    assert (p.read(5) == b'abcde')
 
 def test_read_twice():
     p = Pipeline()
-    p.add("foo")
-    assert (p.read(2) == "fo")
-    assert (p.read(2) == "o")
+    p.add(b"foo")
+    assert (p.read(2) == b"fo")
+    assert (p.read(2) == b"o")
 
 def test_read_twice_empty():
     p = Pipeline()
-    p.add("foo")
-    assert (p.read(2) == "fo")
-    assert (p.read(2) == "o")
-    assert (p.read(2) == "")
+    p.add(b"foo")
+    assert (p.read(2) == b"fo")
+    assert (p.read(2) == b"o")
+    assert (p.read(2) == b"")
 
 def test_read_backup():
     p = Pipeline()
-    p.add("foo")
-    assert (p.read(2) == "fo")
-    p.backup("fo")
-    assert (p.read(2) == "fo")
-    assert (p.read(2) == "o")
+    p.add(b"foo")
+    assert (p.read(2) == b"fo")
+    p.backup(b"fo")
+    assert (p.read(2) == b"fo")
+    assert (p.read(2) == b"o")
 
 def test_read_backup_extra():
     p = Pipeline()
-    p.add("foo")
-    assert (p.read(2) == "fo")
-    p.backup("foobar")
-    assert (p.read(500) == "foobaro")
+    p.add(b"foo")
+    assert (p.read(2) == b"fo")
+    p.backup(b"foobar")
+    assert (p.read(500) == b"foobaro")
 
 def test_read_hybrid_objects():
     p = Pipeline()
-    p.add("foo,")
-    p.add(StringIO("bar,"))
-    p.add(open(FILE))
+    p.add(b"foo,")
+    p.add(BytesIO(b"bar,"))
+    p.add(open(FILE, 'rb'))
 
-    assert (p.read(10) == "foo,bar,#1")
-    assert (p.read(4) == "2345")
-    p.backup("rock") # in the middle of the "file"
-    assert (p.read(6) == "rock67")
+    assert (p.read(10) == b"foo,bar,#1")
+    assert (p.read(4) == b"2345")
+    p.backup(b"rock") # in the middle of the "file"
+    assert (p.read(6) == b"rock67")
 
 def test_close():
     p = Pipeline()
-    p.add("foo")
-    p.add(StringIO("bar"))
+    p.add(b"foo")
+    p.add(BytesIO(b"bar"))
     p.close_request()
-    assert (p.read(1000) == "foobar")
+    assert (p.read(1000) == b"foobar")
     try:
         p.read(1000)
     except PipelineCloseRequest:
@@ -106,25 +106,25 @@ def test_close():
 
 def test_long_1():
     p = Pipeline()
-    p.add("foo")
-    assert (p.read(2) == "fo")
-    p.add("bar")
-    assert (p.read(3) == "oba")
-    p.backup("rocko")
-    p.add(StringIO("soma"))
-    assert (p.read(1000) == "rockorsoma")
-    assert (p.read(1000) == "")
+    p.add(b"foo")
+    assert (p.read(2) == b"fo")
+    p.add(b"bar")
+    assert (p.read(3) == b"oba")
+    p.backup(b"rocko")
+    p.add(BytesIO(b"soma"))
+    assert (p.read(1000) == b"rockorsoma")
+    assert (p.read(1000) == b"")
     assert (p.empty)
-    p.add("X" * 10000)
+    p.add(b"X" * 10000)
     p.close_request()
-    assert (p.read(5000) == 'X' * 5000)
-    p.backup('XXX')
+    assert (p.read(5000) == b'X' * 5000)
+    p.backup(b'XXX')
     try:
-        p.add("newstuff")
+        p.add(b"newstuff")
     except PipelineClosed:
         pass
     assert (not p.empty)
-    assert (p.read(100000) == 'X' * 5003)
+    assert (p.read(100000) == b'X' * 5003)
     assert (not p.empty)
     try:
         p.read(1000)
@@ -138,13 +138,13 @@ def test_long_1():
 
 def test_pri_clean():
     p = Pipeline()
-    p.add("two")
-    p.add("three")
-    p.add("one")
-    assert (p.read(18) == "twothreeone")
+    p.add(b"two")
+    p.add(b"three")
+    p.add(b"one")
+    assert (p.read(18) == b"twothreeone")
 
-    p.add("two", 2)
-    p.add("three", 3)
-    p.add("six", 2)
-    p.add("one", 1)
-    assert (p.read(18) == "threetwosixone")
+    p.add(b"two", 2)
+    p.add(b"three", 3)
+    p.add(b"six", 2)
+    p.add(b"one", 1)
+    assert (p.read(18) == b"threetwosixone")