def send(self, message, timeout=None, tries=3, future=None, callback=None, raw=False):
        '''
        Return a future which will be completed when the message has a response
        '''
        if future is None:
            future = tornado.concurrent.Future()
            future.tries = tries
            future.attempts = 0
            future.timeout = timeout
            # if a future wasn't passed in, we need to serialize the message
            message = self.serial.dumps(message)
        if callback is not None:
            def handle_future(future):
                response = future.result()
                self.io_loop.add_callback(callback, response)
            future.add_done_callback(handle_future)
        # Add this future to the mapping
        self.send_future_map[message] = future

        if self.opts.get('detect_mode') is True:
            timeout = 1

        if timeout is not None:
            send_timeout = self.io_loop.call_later(timeout, self.timeout_message, message)
            self.send_timeout_map[message] = send_timeout

        if not self.send_queue:
            self.io_loop.spawn_callback(self._internal_send_recv)

        self.send_queue.append(message)

        return future