bloom-server.py

import selectors
import socket
import time
from multiprocessing import Process, RawValue, Lock
from bloom_filter2 import BloomFilter
from spamcro.counter import Counter

bind_address = "127.0.0.1"
bind_port = 4500

try:
    import mmap as mmap_mod
except ImportError:
    # Jython lacks mmap()
    HAVE_MMAP = False
else:
    HAVE_MMAP = True

print(f"blf.mmap_available: {HAVE_MMAP}")

class BloomCount:
    added = Counter()
    filtered = Counter()
    reqs = Counter()

blf = BloomFilter(max_elements=150000, error_rate=0.1, filename="./data/blf.bin")
blfc = BloomCount()
_lock = Lock()

sel = selectors.DefaultSelector()

def blf_add(value):
    global blf, blfc, _lock
    blfc.reqs.increment()
    if value in blf:
        blfc.filtered.increment()
        return False
    else:
        with _lock:
            blf.add(value)
        blfc.added.increment()
        return True

def blf_stat():
    global blfc
    return f"c:{str(blfc.added.value())}:f:{str(blfc.filtered.value())}:r:{str(blfc.reqs.value())}"

def accept(sock, mask):
    global blfc
    conn, addr = sock.accept()  # Should be ready
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn, mask):
    global blfc
    data = False
    ttl = time.time() + 20

    while time.time() < ttl:
        try:
            data = conn.recv(1000)  # Should be ready
            print("C:", data)

            if data[0] == 255:
                try:
                    conn.send(f"blf ({blfc.reqs.value()} reqs, {blfc.added.value()} added, {blfc.filtered.value()} filtered)\r\n".encode('utf'))
                    data = conn.recv(1000)  # Should be ready
                    print("C:", data)
                except:
                    data = False
                pass
        except:
            data = False
            pass

        if data:
            added = 0
            count = 0
            arr = data.decode('utf-8').split("\r\n")
            for line in arr:
                if len(line) > 0:
                    count = count + 1
                    if blf_add(line):
                        added = added + 1
                        print(f"~ blf add({len(line)}) '{line}' | +1")
                    else:
                        print(f"~ blf add({len(line)}) '{line}' | -0")
            try:
                if count > 0:
                    conn.send(f"+{added}\r\n".encode('utf-8'))  # Hope it won't block
            except:
                pass
    print('closing', conn)
    sel.unregister(conn)
    conn.close()

def handle_packet(conn, data):
    request_type = zfg.get_request_type(data[0])

    print('zfg.request:', request_type)

    return None

sock = socket.socket()
sock.bind((bind_address, int(bind_port)))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)