nakamuray

nakamuray

import asyncio
import functools
import types


def my_coroutine(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        result_fut = asyncio.Future()

nakamuray

from twisted.internet import defer
from asyncio import futures, get_event_loop

d = defer.Deferred()
d.addCallback(lambda x: x + 1)
d.addCallback(lambda x: x * 2)

f1 = futures.Future()
f2 = futures.Future()
f1.add_done_callback(lambda f: f2.set_result(f.result() + 1))

nakamuray

import asyncio
import aiohttp

from aiohttp import web


class WebsocketProxy(object):
    def __init__(self, upstream_url):
        self.upstream_url = upstream_url

nakamuray

from concurrent.futures import Future

fut = Future()
fut.add_done_callback(lambda f: f.result() * 2)
fut.set_result(21)

print(fut.result())
#assert fut.result() == 42

fut = Future()

nakamuray

'''library to support calling asyncio functions from synchronous code
'''
import asyncio
import functools
import queue
import threading


def create_and_start_loop():
    loop = asyncio.new_event_loop()

nakamuray

var fs = require('fs');

for (i=0; i<10000; i++) {
    fs.writeSync(process.stdout.fd, 'hello world!\n');
}

nakamuray

import t
t.hello

import jedi
script = jedi.Script(path=__file__, line=2, column=7)
definition = script.goto_definitions()[0]
print(repr(definition.docstring()))

nakamuray

import string, sys

t = str.maketrans(string.ascii_uppercase+string.ascii_lowercase,
                  ''.join(chr(u) for u in range(0x1d4d0, 0x1d504)))
print(sys.stdin.read().translate(t))

nakamuray

'''Niconico Style Display

requirements:
    PyQt5
    pyobjc-framework-Cocoa (pyobjc-cocoa?, osx only)
'''

import multiprocessing
import sys
import threading