twitter_clone_pyredis.py

import http.server
import cgi
import functools
import json
import math
import random
import socket
import socketserver
import time
import threading
import unittest
import uuid
import urllib.parse

import redis


def to_bytes(x):
    return x.encode() if isinstance(x, str) else x


def to_str(x):
    return x.decode() if isinstance(x, bytes) else x


def acquire_lock_with_timeout(
    conn, lockname, acquire_timeout=10, lock_timeout=10):
    identifier = str(uuid.uuid4())  # A
    lockname = 'lock:' + lockname
    lock_timeout = int(math.ceil(lock_timeout))  # D

    end = time.time() + acquire_timeout
    while time.time() < end:
        if conn.setnx(lockname, identifier):  # B
            conn.expire(lockname, lock_timeout)  # B
            return identifier
        elif conn.ttl(lockname) < 0:  # C
            conn.expire(lockname, lock_timeout)  # C

        time.sleep(.001)

    return False


def release_lock(conn, lockname, identifier):
    pipe = conn.pipeline(True)
    lockname = 'lock:' + lockname
    identifier = to_bytes(identifier)

    while True:
        try:
            pipe.watch(lockname)  # A
            if pipe.get(lockname) == identifier:  # A
                pipe.multi()  # B
                pipe.delete(lockname)  # B
                pipe.execute()  # B
                return True  # B

            pipe.unwatch()
            break

        except redis.exceptions.WatchError:  # C
            pass  # C

    return False  # D


CONFIGS = {}
CHECKED = {}


def get_config(conn, type, component, wait=1):
    key = 'config:%s:%s' % (type, component)

    if CHECKED.get(key) < time.time() - wait:  # A
        CHECKED[key] = time.time()  # B
        config = json.loads(conn.get(key) or '{}')  # C
        old_config = CONFIGS.get(key)  # D

        if config != old_config:  # E
            CONFIGS[key] = config  # F

    return CONFIGS.get(key)


REDIS_CONNECTIONS = {}


def redis_connection(component, wait=1):  # A
    key = 'config:redis:' + component  # B

    def wrapper(function):  # C
        @functools.wraps(function)  # D
        def call(*args, **kwargs):  # E
            old_config = CONFIGS.get(key, object())  # F
            _config = get_config(  # G
                config_connection, 'redis', component, wait)  # G

            config = {}
            for k, v in _config.items():  # L
                config[k.encode('utf-8')] = v  # L

            if config != old_config:  # H
                REDIS_CONNECTIONS[key] = redis.Redis(**config)  # H

            return function(  # I
                REDIS_CONNECTIONS.get(key), *args, **kwargs)  # I

        return call  # J

    return wrapper  # K


def execute_later(conn, queue, name, args):
    # this is just for testing purposes
    assert conn is args[0]
    t = threading.Thread(target=globals()[name], args=tuple(args))
    t.setDaemon(1)
    t.start()


# <start id="create-twitter-user"/>
def create_user(conn, login, name):
    llogin = login.lower()
    lock = acquire_lock_with_timeout(conn, 'user:' + llogin, 1)  # A
    if not lock:  # B
        return None  # B

    if conn.hget('users:', llogin):  # C
        release_lock(conn, 'user:' + llogin, lock)  # C
        return None  # C

    id = conn.incr('user:id:')  # D
    pipeline = conn.pipeline(True)
    pipeline.hset('users:', llogin, id)  # E
    pipeline.hmset('user:%s' % id, {  # F
        'login': login,  # F
        'id': id,  # F
        'name': name,  # F
        'followers': 0,  # F
        'following': 0,  # F
        'posts': 0,  # F
        'signup': time.time(),  # F
    })
    pipeline.execute()
    release_lock(conn, 'user:' + llogin, lock)  # G
    return id  # H


# <end id="create-twitter-user"/>
# A Try to acquire the lock for the lowercased version of the login name. This function is defined in chapter 6
# B If we couldn't get the lock, then someone else already has the same login name
# C We also store a HASH of lowercased login names to user ids, so if there is already a login name that maps to an ID, we know and won't give it to a second person
# D Each user is given a unique id, generated by incrementing a counter
# E Add the lowercased login name to the HASH that maps from login names to user ids
# F Add the user information to the user's HASH
# G Release the lock over the login name
# H Return the id of the user
# END

# <start id="create-twitter-status"/>
def create_status(conn, uid, message, **data):
    pipeline = conn.pipeline(True)
    pipeline.hget('user:%s' % uid, 'login')  # A
    pipeline.incr('status:id:')  # B
    login, id = pipeline.execute()

    if not login:  # C
        return None  # C

    data.update({
        'message': message,  # D
        'posted': time.time(),  # D
        'id': id,  # D
        'uid': uid,  # D
        'login': login,  # D
    })
    pipeline.hmset('status:%s' % id, data)  # D
    pipeline.hincrby('user:%s' % uid, 'posts')  # E
    pipeline.execute()
    return id  # F


# <end id="create-twitter-status"/>
# A Get the user's login name from their user id
# B Create a new id for the status message
# C Verify that we have a proper user account before posting
# D Prepare and set the data for the status message
# E Record the fact that a status message has been posted
# F Return the id of the newly created status message
# END

# <start id="fetch-page"/>
def get_status_messages(conn, uid, timeline='home:', page=1, count=30):  # A
    statuses = conn.zrevrange(  # B
        '%s%s' % (timeline, uid), (page - 1) * count, page * count - 1)  # B

    pipeline = conn.pipeline(True)
    for id in statuses:  # C
        pipeline.hgetall('status:%s' % (to_str(id),))  # C

    return [_f for _f in pipeline.execute() if _f]  # D


# <end id="fetch-page"/>
# A We will take an optional 'timeline' argument, as well as page size and status message counts
# B Fetch the most recent status ids in the timeline
# C Actually fetch the status messages themselves
# D Filter will remove any 'missing' status messages that had been previously deleted
# END

# <start id="follow-user"/>
HOME_TIMELINE_SIZE = 1000


def follow_user(conn, uid, other_uid):
    fkey1 = 'following:%s' % uid  # A
    fkey2 = 'followers:%s' % other_uid  # A

    if conn.zscore(fkey1, other_uid):  # B
        return None  # B

    now = time.time()

    pipeline = conn.pipeline(True)
    pipeline.zadd(fkey1, {other_uid: now})  # C
    pipeline.zadd(fkey2, {uid: now})  # C
    pipeline.zrevrange('profile:%s' % other_uid,  # E
                       0, HOME_TIMELINE_SIZE - 1, withscores=True)  # E
    following, followers, status_and_score = pipeline.execute()[-3:]

    pipeline.hincrby('user:%s' % uid, 'following', int(following))  # F
    pipeline.hincrby('user:%s' % other_uid, 'followers', int(followers))  # F
    if status_and_score:
        pipeline.zadd('home:%s' % uid, dict(status_and_score))  # G
    pipeline.zremrangebyrank('home:%s' % uid, 0, -HOME_TIMELINE_SIZE - 1)  # G

    pipeline.execute()
    return True  # H


# <end id="follow-user"/>
# A Cache the following and followers key names
# B If the other_uid is already being followed, return
# C Add the uids to the proper following and followers ZSETs
# E Fetch the most recent HOME_TIMELINE_SIZE status messages from the newly followed user's profile timeline
# F Update the known size of the following and followers ZSETs in each user's HASH
# G Update the home timeline of the following user, keeping only the most recent 1000 status messages
# H Return that the user was correctly followed
# END

# <start id="unfollow-user"/>
def unfollow_user(conn, uid, other_uid):
    fkey1 = 'following:%s' % uid  # A
    fkey2 = 'followers:%s' % other_uid  # A

    if not conn.zscore(fkey1, other_uid):  # B
        return None  # B

    pipeline = conn.pipeline(True)
    pipeline.zrem(fkey1, other_uid)  # C
    pipeline.zrem(fkey2, uid)  # C
    pipeline.zrevrange('profile:%s' % other_uid,  # E
                       0, HOME_TIMELINE_SIZE - 1)  # E
    following, followers, statuses = pipeline.execute()[-3:]

    pipeline.hincrby('user:%s' % uid, 'following', -int(following))  # F
    pipeline.hincrby('user:%s' % other_uid, 'followers', -int(followers))  # F
    if statuses:
        pipeline.zrem('home:%s' % uid, *statuses)  # G

    pipeline.execute()
    return True  # H


# <end id="unfollow-user"/>
# A Cache the following and followers key names
# B If the other_uid is not being followed, return
# C Remove the uids the proper following and followers ZSETs
# E Fetch the most recent HOME_TIMELINE_SIZE status messages from the user that we stopped following
# F Update the known size of the following and followers ZSETs in each user's HASH
# G Update the home timeline, removing any status messages from the previously followed user
# H Return that the unfollow executed successfully
# END

# <start id="exercise-refilling-timelines"/>
REFILL_USERS_STEP = 50


def refill_timeline(conn, incoming, timeline, start=0):
    if not start and conn.zcard(timeline) >= 750:  # A
        return  # A

    users = conn.zrangebyscore(incoming, start, 'inf',  # B
                               start=0, num=REFILL_USERS_STEP, withscores=True)  # B

    pipeline = conn.pipeline(False)
    for uid, start in users:
        uid = to_str(uid)
        pipeline.zrevrange('profile:%s' % uid,  # C
                           0, HOME_TIMELINE_SIZE - 1, withscores=True)  # C

    messages = []
    for results in pipeline.execute():
        messages.extend(results)  # D

    messages.sort(key=lambda x: -x[1])  # E
    del messages[HOME_TIMELINE_SIZE:]  # E

    pipeline = conn.pipeline(True)
    if messages:
        pipeline.zadd(timeline, dict(messages))  # F
    pipeline.zremrangebyrank(  # G
        timeline, 0, -HOME_TIMELINE_SIZE - 1)  # G
    pipeline.execute()

    if len(users) >= REFILL_USERS_STEP:
        execute_later(conn, 'default', 'refill_timeline',  # H
                      [conn, incoming, timeline, start])  # H


# <end id="exercise-refilling-timelines"/>
# A If the timeline is 3/4 of the way full already, don't bother refilling it
# B Fetch a group of users that should contribute to this timeline
# C Fetch the most recent status messages from the users followed
# D Group all of the fetched status messages together
# E Sort all of the status messages by how recently they were posted, and keep the most recent 1000
# F Add all of the fetched status messages to the user's home timeline
# G Remove any messages that are older than the most recent 1000
# H If there are still more users left to fetch from, keep going
# END

# <start id="exercise-follow-user-list"/>
def follow_user_list(conn, other_uid, list_id):
    fkey1 = 'list:in:%s' % list_id  # A
    fkey2 = 'list:out:%s' % other_uid  # A
    timeline = 'list:statuses:%s' % list_id  # A

    if conn.zscore(fkey1, other_uid):  # B
        return None  # B

    now = time.time()

    pipeline = conn.pipeline(True)
    pipeline.zadd(fkey1, {other_uid: now})  # C
    pipeline.zadd(fkey2, {list_id: now})  # C
    pipeline.zrevrange('profile:%s' % other_uid,  # D
                       0, HOME_TIMELINE_SIZE - 1, withscores=True)  # D
    following, followers, status_and_score = pipeline.execute()[-3:]

    pipeline.hincrby('list:%s' % list_id, 'following', int(following))  # E
    pipeline.zadd(timeline, dict(status_and_score))  # F
    pipeline.zremrangebyrank(timeline, 0, -HOME_TIMELINE_SIZE - 1)  # F

    pipeline.execute()
    return True  # G


# <end id="exercise-follow-user"/>
# A Cache the key names
# B If the other_uid is already being followed by the list, return
# C Add the uids to the proper ZSETs
# D Fetch the most recent status messages from the user's profile timeline
# E Update the known size of the list ZSETs in the list information HASH
# F Update the list of status messages
# G Return that adding the user to the list completed successfully
# END

# <start id="exercise-unfollow-user-list"/>
def unfollow_user_list(conn, other_uid, list_id):
    fkey1 = 'list:in:%s' % list_id  # A
    fkey2 = 'list:out:%s' % other_uid  # A
    timeline = 'list:statuses:%s' % list_id  # A

    if not conn.zscore(fkey1, other_uid):  # B
        return None  # B

    pipeline = conn.pipeline(True)
    pipeline.zrem(fkey1, other_uid)  # C
    pipeline.zrem(fkey2, list_id)  # C
    pipeline.zrevrange('profile:%s' % other_uid,  # D
                       0, HOME_TIMELINE_SIZE - 1)  # D
    following, followers, statuses = pipeline.execute()[-3:]

    pipeline.hincrby('list:%s' % list_id, 'following', -int(following))  # E
    if statuses:
        pipeline.zrem(timeline, *statuses)  # F
        refill_timeline(fkey1, timeline)  # G

    pipeline.execute()
    return True  # H


# <end id="exercise-unfollow-user-list"/>
# A Cache the key names
# B If the other_uid is not being followed by the list, return
# C Remove the uids from the proper ZSETs
# D Find the size of the list ZSET
# D Fetch the most recent status messages from the user that we stopped following
# E Update the known size of the list ZSETs in the list information HASH
# F Update the list timeline, removing any status messages from the previously followed user
# G Start refilling the list timeline
# H Return that the unfollow executed successfully
# END

# <start id="exercise-create-user-list"/>
def create_user_list(conn, uid, name):
    pipeline = conn.pipeline(True)
    pipeline.hget('user:%s' % uid, 'login')  # A
    pipeline.incr('list:id:')  # B
    login, id = pipeline.execute()

    if not login:  # C
        return None  # C

    now = time.time()

    pipeline = conn.pipeline(True)
    pipeline.zadd('lists:%s' % uid, {id: now})  # D
    pipeline.hmset('list:%s' % id, {  # E
        'name': name,  # E
        'id': id,  # E
        'uid': uid,  # E
        'login': login,  # E
        'following': 0,  # E
        'created': now,  # E
    })
    pipeline.execute()

    return id  # F


# <end id="exercise-create-user-list"/>
# A Fetch the login name of the user who is creating the list
# B Generate a new list id
# C If the user doesn't exist, return
# D Add the new list to a ZSET of lists that the user has created
# E Create the list information HASH
# F Return the new list id
# END

# <start id="post-message"/>
def post_status(conn, uid, message, **data):
    id = create_status(conn, uid, message, **data)  # A
    if not id:  # B
        return None  # B

    posted = conn.hget('status:%s' % id, 'posted')  # C
    if not posted:  # D
        return None  # D

    post = {str(id): float(posted)}
    conn.zadd('profile:%s' % uid, post)  # E

    syndicate_status(conn, uid, post)  # F
    return id


# <end id="post-message"/>
# A Create a status message using the earlier function
# B If the creation failed, return
# C Get the time that the message was posted
# D If the post wasn't found, return
# E Add the status message to the user's profile timeline
# F Actually push the status message out to the followers of the user
# END

# <start id="syndicate-message"/>
POSTS_PER_PASS = 1000  # A


def syndicate_status(conn, uid, post, start=0):
    followers = conn.zrangebyscore('followers:%s' % uid, start, 'inf',  # B
                                   start=0, num=POSTS_PER_PASS, withscores=True)  # B

    pipeline = conn.pipeline(False)
    for follower, start in followers:  # E
        follower = to_str(follower)
        pipeline.zadd('home:%s' % follower, post)  # C
        pipeline.zremrangebyrank(  # C
            'home:%s' % follower, 0, -HOME_TIMELINE_SIZE - 1)  # C
    pipeline.execute()

    if len(followers) >= POSTS_PER_PASS:  # D
        execute_later(conn, 'default', 'syndicate_status',  # D
                      [conn, uid, post, start])  # D


# <end id="syndicate-message"/>
# A Only send to 1000 users per pass
# B Fetch the next group of 1000 followers, starting at the last person to be updated last time
# E Iterating through the followers results will update the 'start' variable, which we can later pass on to subsequent syndicate_status() calls
# C Add the status to the home timelines of all of the fetched followers, and trim the home timelines so they don't get too big
# D If at least 1000 followers had received an update, execute the remaining updates in a task
# END

# <start id="syndicate-message-list"/>
def syndicate_status_list(conn, uid, post, start=0, on_lists=False):
    key = 'followers:%s' % uid  # A
    base = 'home:%s'  # A
    if on_lists:  # A
        key = 'list:out:%s' % uid  # A
        base = 'list:statuses:%s'  # A
    followers = conn.zrangebyscore(key, start, 'inf',  # B
                                   start=0, num=POSTS_PER_PASS, withscores=True)  # B

    pipeline = conn.pipeline(False)
    for follower, start in followers:  # C
        follower = to_str(follower)
        pipeline.zadd(base % follower, post)  # C
        pipeline.zremrangebyrank(  # C
            base % follower, 0, -HOME_TIMELINE_SIZE - 1)  # C
    pipeline.execute()

    if len(followers) >= POSTS_PER_PASS:  # D
        execute_later(conn, 'default', 'syndicate_status',  # D
                      [conn, uid, post, start, on_lists])  # D

    elif not on_lists:
        execute_later(conn, 'default', 'syndicate_status',  # E
                      [conn, uid, post, 0, True])  # E


# <end id="syndicate-message-list"/>
# A Use keys for home timelines or list timelines, depending on how far along we are
# B Fetch the next group of 1000 followers or lists, starting at the last user or list to be updated last time
# C Add the status to the home timelines of all of the fetched followers, and trim the home timelines so they don't get too big
# D If at least 1000 followers had received an update, execute the remaining updates in a task
# E Start executing over lists if we haven't executed over lists yet, but we are done with home timelines
# END

# <start id="delete-message"/>
def delete_status(conn, uid, status_id):
    status_id = to_str(status_id)
    key = 'status:%s' % status_id
    lock = acquire_lock_with_timeout(conn, key, 1)  # A
    if not lock:  # B
        return None  # B

    if conn.hget(key, 'uid') != to_bytes(uid):  # C
        release_lock(conn, key, lock)  # C
        return None  # C

    uid = to_str(uid)
    pipeline = conn.pipeline(True)
    pipeline.delete(key)  # D
    pipeline.zrem('profile:%s' % uid, status_id)  # E
    pipeline.zrem('home:%s' % uid, status_id)  # F
    pipeline.hincrby('user:%s' % uid, 'posts', -1)  # G
    pipeline.execute()

    release_lock(conn, key, lock)
    return True


# <end id="delete-message"/>
# A Acquire a lock around the status object to ensure that no one else is trying to delete it when we are
# B If we didn't get the lock, return
# C If the user doesn't match the user stored in the status message, return
# D Delete the status message
# E Remove the status message id from the user's profile timeline
# F Remove the status message id from the user's home timeline
# G Reduce the number of posted messages in the user information HASH
# END

# <start id="exercise-clean-out-timelines"/>
def clean_timelines(conn, uid, status_id, start=0, on_lists=False):
    uid = to_str(uid)
    status_id = to_str(status_id)
    key = 'followers:%s' % uid  # A
    base = 'home:%s'  # A
    if on_lists:  # A
        key = 'list:out:%s' % uid  # A
        base = 'list:statuses:%s'  # A
    followers = conn.zrangebyscore(key, start, 'inf',  # B
                                   start=0, num=POSTS_PER_PASS, withscores=True)  # B

    pipeline = conn.pipeline(False)
    for follower, start in followers:  # C
        follower = to_str(follower)
        pipeline.zrem(base % follower, status_id)  # C
    pipeline.execute()

    if len(followers) >= POSTS_PER_PASS:  # D
        execute_later(conn, 'default', 'clean_timelines',  # D
                      [conn, uid, status_id, start, on_lists])  # D

    elif not on_lists:
        execute_later(conn, 'default', 'clean_timelines',  # E
                      [conn, uid, status_id, 0, True])  # E


# <end id="exercise-clean-out-timelines"/>
# A Use keys for home timelines or list timelines, depending on how far along we are
# B Fetch the next group of 1000 followers or lists, starting at the last user or list to be updated last time
# C Remove the status from the home timelines of all of the fetched followers
# D If at least 1000 followers had received an update, execute the remaining updates in a task
# E Start executing over lists if we haven't executed over lists yet, but we are done with home timelines
# END

# <start id="streaming-http-server"/>
class StreamingAPIServer(  # A
    socketserver.ThreadingMixIn,  # B
    http.server.HTTPServer):  # B

    daemon_threads = True  # C


class StreamingAPIRequestHandler(  # D
    http.server.BaseHTTPRequestHandler):  # E

    def do_GET(self):  # F
        parse_identifier(self)  # G
        if self.path != '/statuses/sample.json':  # H
            return self.send_error(404)  # H

        process_filters(self)  # I

    def do_POST(self):  # J
        parse_identifier(self)  # K
        if self.path != '/statuses/filter.json':  # L
            return self.send_error(404)  # L

        process_filters(self)  # M


# <end id="streaming-http-server"/>
# A Create a new class called 'StreamingAPIServer'
# B This new class should have the ability to create new threads with each request, and should be a HTTPServer
# C Tell the internals of the threading server to shut down all client request threads if the main server thread dies
# D Create a new class called 'StreamingAPIRequestHandler'
# E This new class should be able to handle HTTP requests
# F Create a method that is called do_GET(), which will be executed on GET requests performed against this server
# G Call a helper function that handles the fetching of an identifier for the client
# H If the request is not a 'sample' or 'firehose' streaming GET request, return a '404 not found' error
# I Otherwise, call a helper function that actually handles the filtering
# J Create a method that is called do_POST(), which will be executed on POST requests performed against this server
# K Call a helper function that handles the fetching of an identifier for the client
# L If the request is not a user, keyword, or location filter, return a '404 not found' error
# M Otherwise, call a helper function that actually handles the filtering
# END

# <start id="get-identifier"/>
def parse_identifier(handler):
    handler.identifier = None  # A
    handler.query = {}  # A
    if '?' in handler.path:  # B
        handler.path, _, query = handler.path.partition('?')  # C
        handler.query = urllib.parse.parse_qs(query)  # D
        identifier = handler.query.get('identifier') or [None]  # E
        handler.identifier = identifier[0]  # F


# <end id="get-identifier"/>
# A Set the identifier and query arguments to be palceholder values
# B If there were query arguments as part of the request, process them
# C Extract the query portion from the path, and update the path
# D Parse the query
# E Fetch the list of query arguments with the name 'identifier'
# F Use the first identifier passed
# END

# <start id="stream-to-client"/>
FILTERS = ('track', 'filter', 'location')  # A


def process_filters(handler):
    id = handler.identifier
    if not id:  # B
        return handler.send_error(401, "identifier missing")  # B

    method = handler.path.rsplit('/')[-1].split('.')[0]  # C
    name = None
    args = None
    if method == 'filter':  # D
        data = cgi.FieldStorage(  # E
            fp=handler.rfile,  # E
            headers=handler.headers,  # E
            environ={'REQUEST_METHOD': 'POST',  # E
                     'CONTENT_TYPE': handler.headers['Content-Type'],  # E
                     })

        for name in data:  # F
            if name in FILTERS:  # F
                args = data.getfirst(name).lower().split(',')  # F
                break  # F

        if not args:  # G
            return handler.send_error(401, "no filter provided")  # G
    else:
        args = handler.query  # M

    handler.send_response(200)  # H
    handler.send_header('Transfer-Encoding', 'chunked')  # H
    handler.end_headers()

    quit = [False]  # N
    for item in filter_content(id, method, name, args, quit):  # I
        try:
            handler.wfile.write('%X\r\n%s\r\n' % (len(item), item))  # J
        except socket.error:  # K
            quit[0] = True  # K
    if not quit[0]:
        handler.wfile.write('0\r\n\r\n')  # L


# <end id="stream-to-client"/>
# A Keep a listing of filters that need arguments
# B Return an error if an identifier was not provided by the client
# C Fetch the method, should be one of 'sample' or 'filter'
# D If this is a filtering method, we need to fetch the arguments
# E Parse the POST request to discover the type and arguments to the filter
# F Fetch any of the filters provided by the client request
# G If there were no filters specified, return an error
# M For sample requests, pass the query arguments as the 'args'
# H Finally return a response to the client, informing them that they will be receiving a streaming response
# N Use a Python list as a holder for a pass-by-reference variable, which will allow us to tell the content filter to stop receiving messages
# I Iterate over the results of the filter
# J Send the pre-encoded response to the client using the chunked encoding
# K If sending to the client caused an error, then we need to tell the subscriber to unsubscribe and shut down
# L Send the "end of chunks" message to the client if we haven't already disconnected
# END

_create_status = create_status


# <start id="create-message-streaming"/>
def create_status(conn, uid, message, **data):
    pipeline = conn.pipeline(True)
    pipeline.hget('user:%s' % uid, 'login')
    pipeline.incr('status:id:')
    login, id = pipeline.execute()

    if not login:
        return None

    data.update({
        'message': message,
        'posted': time.time(),
        'id': id,
        'uid': uid,
        'login': to_str(login),
    })
    pipeline.hmset('status:%s' % id, data)
    pipeline.hincrby('user:%s' % uid, 'posts')
    pipeline.publish('streaming:status:', json.dumps(data))  # A
    pipeline.execute()
    return id


# <end id="create-message-streaming"/>
# A The added line to send a message to streaming filters
# END

_delete_status = delete_status


# <start id="delete-message-streaming"/>
def delete_status(conn, uid, status_id):
    # raise Exception("what the fuck")
    status_id = to_str(status_id)
    key = 'status:%s' % status_id
    lock = acquire_lock_with_timeout(conn, key, 1)
    if not lock:
        return None

    if conn.hget(key, 'uid') != to_bytes(uid):
        release_lock(conn, key, lock)
        return None

    uid = to_str(uid)
    pipeline = conn.pipeline(True)
    status = conn.hgetall(key)  # A
    status = {to_str(k): to_str(v) for k, v in status.items()}
    status['deleted'] = True  # B
    pipeline.publish('streaming:status:', json.dumps(status))  # C
    pipeline.delete(key)
    pipeline.zrem('profile:%s' % uid, status_id)
    pipeline.zrem('home:%s' % uid, status_id)
    pipeline.hincrby('user:%s' % uid, 'posts', -1)
    pipeline.execute()

    release_lock(conn, key, lock)
    return True


# <end id="delete-message-streaming"/>
# A Fetch the status message so that streaming filters can perform the same filters to determine whether the deletion should be passed to the client
# B Mark the status message as deleted
# C Publish the deleted status message to the stream
# END

# <start id="message-subscription"/>
@redis_connection('social-network')  # A
def filter_content(conn, id, method, name, args, quit):
    match = create_filters(id, method, name, args)  # B

    pubsub = conn.pubsub()  # C
    pubsub.subscribe(['streaming:status:'])  # C

    for item in pubsub.listen():  # D
        message = item['data']  # E
        decoded = json.loads(message)  # E

        if match(decoded):  # F
            if decoded.get('deleted'):  # G
                yield json.dumps({  # G
                    'id': decoded['id'], 'deleted': True})  # G
            else:
                yield message  # H

        if quit[0]:  # I
            break  # I

    pubsub.reset()  # J


# <end id="message-subscription"/>
# A Use our automatic connection decorator from chapter 5
# B Create the filter that will determine whether a message should be sent to the client
# C Prepare the subscription
# D Receive messages from the subscription
# E Get the status message information from the subscription structure
# F Check if the status message matched the filter
# G For deleted messages, send a special 'deleted' placeholder for the message
# H For matched status messages that are not deleted, send the message itself
# I If the web server no longer has a connection to the client, stop filtering messages
# J Reset the Redis connection to ensure that the Redis server clears its outgoing buffers if this wasn't fast enough
# END

# <start id="create-filters"/>
def create_filters(id, method, name, args):
    if method == 'sample':  # A
        return SampleFilter(id, args)  # A
    elif name == 'track':  # B
        return TrackFilter(args)  # B
    elif name == 'follow':  # B
        return FollowFilter(args)  # B
    elif name == 'location':  # B
        return LocationFilter(args)  # B
    raise Exception("Unknown filter")  # C


# <end id="create-filters"/>
# A For the 'sample' method, we don't need to worry about names, just the arguments
# B For the 'filter' method, we actually worry about which of the filters we want to apply, so return the specific filters for them
# C If no filter matches, then raise an exception
# END

# <start id="sample-filter"/>
def SampleFilter(id, args):  # A
    percent = int(args.get('percent', ['10'])[0], 10)  # B
    ids = list(range(100))  # C
    shuffler = random.Random(id)  # C
    shuffler.shuffle(ids)  # C
    keep = set(ids[:max(percent, 1)])  # D

    def check(status):  # E
        return (status['id'] % 100) in keep  # F

    return check


# <end id="sample-filter"/>
# A We are defining a filter class called "SampleFilter", which are created by passing 'id' and 'args' parameters
# B The 'args' parameter is actually a dictionary, based on the parameters passed as part of the GET request
# C We use the 'id' parameter to randomly choose a subset of ids, the count of which is determined by the 'percent' argument passed
# D We will use a Python set to allow us to quickly determine whether a status message matches our criteria
# E If we create a specially named method called '__call__' on an instance, it will be called if the instance is used like a function
# F To filter status messages, we fetch the status id, find its value modulo 100, and return whether it is in the status ids that we want to accept
# END

# <start id="track-filter"/>
def TrackFilter(list_of_strings):
    groups = []  # A
    for group in list_of_strings:  # A
        group = set(group.lower().split())  # A
        if group:
            groups.append(group)  # B

    def check(status):
        message_words = set(status['message'].lower().split())  # C
        for group in groups:  # D
            if len(group & message_words) == len(group):  # E
                return True  # E
        return False

    return check


# <end id="track-filter"/>
# A The filter should have been provided with a list of word groups, and the filter matches if a message has all of the words in any of the groups
# B We will only keep groups that have at least 1 word
# C We are going to split words in the message on whitespace
# D Then we are going to iterate over all of the groups
# E If all of the words in any of the groups match, we will accept the message with this filter
# END

# <start id="follow-filter"/>
def FollowFilter(names):
    nset = set()  # A
    for name in names:  # B
        nset.add('@' + name.lower().lstrip('@'))  # B

    def check(status):
        message_words = set(status['message'].lower().split())  # C
        message_words.add('@' + status['login'].lower())  # C

        return message_words & nset  # D

    return check


# <end id="follow-filter"/>
# A We are going to try to match login names against posters and messages
# B Make all of the names consistently stored as '@username'
# C Construct a set of words from the message and the poster's name
# D Consider the message a match if any of the usernames provided match any of the whitespace-separated words in the message
# END

# <start id="location-filter"/>
def LocationFilter(list_of_boxes):
    boxes = []  # A
    for start in range(0, len(list_of_boxes) - 3, 4):  # A
        boxes.append(list(map(float, list_of_boxes[start:start + 4])))  # A

    def check(self, status):
        location = status.get('location')  # B
        if not location:  # C
            return False  # C

        lat, lon = list(map(float, location.split(',')))  # D
        for box in self.boxes:  # E
            if (box[1] <= lat <= box[3] and  # F
                box[0] <= lon <= box[2]):  # F
                return True  # F
        return False

    return check


# <end id="location-filter"/>
# A We are going to create a set of boxes that define the regions that should return messages
# B Try to fetch 'location' data from a status message
# C If the message has no location information, then it can't be inside the boxes
# D Otherwise, extract the latitude and longitude of the location
# E To match one of the boxes, we need to iterate over all boxes
# F If the message status location is within the required latitude and longitude range, then the status message matches the filter
# END

_filter_content = filter_content


def filter_content(identifier, method, name, args, quit):
    print("got:", identifier, method, name, args)
    for i in range(10):
        yield json.dumps({'id': i})
        if quit[0]:
            break
        time.sleep(.1)


'''
# <start id="start-http-server"/>
if __name__ == '__main__':                  #A
    server = StreamingAPIServer(                        #B
        ('localhost', 8080), StreamingAPIRequestHandler)#B
    print 'Starting server, use <Ctrl-C> to stop'       #C
    server.serve_forever()                  #D
# <end id="start-http-server"/>
#A Run the below block of code if this module is being run from the command line
#B Create an insteance of the streaming API server listening on localhost port 8080, and use the StreamingAPIRequestHandler to process requests
#C Print an informational line
#D Run the server until someone kills it
#END
'''


class TestCh08(unittest.TestCase):
    def setUp(self):
        self.conn = redis.Redis(db=15)
        self.conn.flushdb()

    def tearDown(self):
        self.conn.flushdb()

    def test_create_user_and_status(self):
        self.assertEqual(create_user(self.conn, 'TestUser', 'Test User'), 1)
        self.assertEqual(create_user(self.conn, 'TestUser', 'Test User2'), None)

        self.assertEqual(create_status(self.conn, 1, "This is a new status message"), 1)
        self.assertEqual(self.conn.hget('user:1', 'posts'), b'1')

    def test_follow_unfollow_user(self):
        self.assertEqual(create_user(self.conn, 'TestUser', 'Test User'), 1)
        self.assertEqual(create_user(self.conn, 'TestUser2', 'Test User2'), 2)

        self.assertTrue(follow_user(self.conn, 1, 2))
        self.assertEqual(self.conn.zcard('followers:2'), 1)
        self.assertEqual(self.conn.zcard('followers:1'), 0)
        self.assertEqual(self.conn.zcard('following:1'), 1)
        self.assertEqual(self.conn.zcard('following:2'), 0)
        self.assertEqual(self.conn.hget('user:1', 'following'), b'1')
        self.assertEqual(self.conn.hget('user:2', 'following'), b'0')
        self.assertEqual(self.conn.hget('user:1', 'followers'), b'0')
        self.assertEqual(self.conn.hget('user:2', 'followers'), b'1')

        self.assertEqual(unfollow_user(self.conn, 2, 1), None)
        self.assertEqual(unfollow_user(self.conn, 1, 2), True)
        self.assertEqual(self.conn.zcard('followers:2'), 0)
        self.assertEqual(self.conn.zcard('followers:1'), 0)
        self.assertEqual(self.conn.zcard('following:1'), 0)
        self.assertEqual(self.conn.zcard('following:2'), 0)
        self.assertEqual(self.conn.hget('user:1', 'following'), b'0')
        self.assertEqual(self.conn.hget('user:2', 'following'), b'0')
        self.assertEqual(self.conn.hget('user:1', 'followers'), b'0')
        self.assertEqual(self.conn.hget('user:2', 'followers'), b'0')

    def test_syndicate_status(self):
        self.assertEqual(create_user(self.conn, 'TestUser', 'Test User'), 1)
        self.assertEqual(create_user(self.conn, 'TestUser2', 'Test User2'), 2)
        self.assertTrue(follow_user(self.conn, 1, 2))
        self.assertEqual(self.conn.zcard('followers:2'), 1)
        self.assertEqual(self.conn.hget('user:1', 'following'), b'1')
        self.assertEqual(post_status(self.conn, 2, 'this is some message content'), 1)
        self.assertEqual(len(get_status_messages(self.conn, 1)), 1)

        for i in range(3, 11):
            self.assertEqual(create_user(self.conn, 'TestUser%s' % i, 'Test User%s' % i), i)
            follow_user(self.conn, i, 2)

        global POSTS_PER_PASS
        POSTS_PER_PASS = 5

        self.assertEqual(post_status(self.conn, 2, 'this is some other message content'), 2)
        time.sleep(.1)
        self.assertEqual(len(get_status_messages(self.conn, 9)), 2)

        self.assertTrue(unfollow_user(self.conn, 1, 2))
        self.assertEqual(len(get_status_messages(self.conn, 1)), 0)

    def test_refill_timeline(self):
        self.assertEqual(create_user(self.conn, 'TestUser', 'Test User'), 1)
        self.assertEqual(create_user(self.conn, 'TestUser2', 'Test User2'), 2)
        self.assertEqual(create_user(self.conn, 'TestUser3', 'Test User3'), 3)

        self.assertTrue(follow_user(self.conn, 1, 2))
        self.assertTrue(follow_user(self.conn, 1, 3))

        global HOME_TIMELINE_SIZE
        HOME_TIMELINE_SIZE = 5

        for i in range(10):
            self.assertTrue(post_status(self.conn, 2, 'message'))
            self.assertTrue(post_status(self.conn, 3, 'message'))
            time.sleep(.05)

        self.assertEqual(len(get_status_messages(self.conn, 1)), 5)
        self.assertTrue(unfollow_user(self.conn, 1, 2))
        self.assertTrue(len(get_status_messages(self.conn, 1)) < 5)

        refill_timeline(self.conn, 'following:1', 'home:1')
        messages = get_status_messages(self.conn, 1)
        self.assertEqual(len(messages), 5)
        for msg in messages:
            self.assertEqual(msg[b'uid'], b'3')

        delete_status(self.conn, '3', messages[-1][b'id'])
        self.assertEqual(len(get_status_messages(self.conn, 1)), 4)
        self.assertEqual(self.conn.zcard('home:1'), 5)
        clean_timelines(self.conn, '3', messages[-1][b'id'])
        self.assertEqual(self.conn.zcard('home:1'), 4)


if __name__ == '__main__':
    unittest.main()