Add experimental 'postd' implementation.

git-svn-id: http://opensvn.csie.org/pttbbs/trunk@6027 63ad8ddf-47c3-0310-b6dd-a9e9d9715204

4 files changed, 218 insertions, 9 deletions

diff --git a/pttbbs/daemon/postd/postd.py b/pttbbs/daemon/postd/postd.py
new file mode 100755
index 00000000..bc159caf
--- /dev/null
+++ b/pttbbs/daemon/postd/postd.py
@@ -0,0 +1,147 @@
+#!/usr/bin/env python
+
+import collections
+import json
+import logging
+import os
+import struct
+import sys
+
+from gevent import monkey; monkey.patch_all()
+import gevent
+import gevent.server
+import leveldb
+
+from pyutil import pttstruct
+from pyutil import big5
+
+# Ref: ../../include/daemons.h
+RequestFormatString = 'HH'
+Request = collections.namedtuple('Request', 'cb operation')
+PostKeyFormatString = '%ds%ds' % (pttstruct.IDLEN + 1, pttstruct.FNLEN + 1)
+PostKey = collections.namedtuple('PostKey', 'board file')
+# TODO in future we don't need userref, and ctime should be automatically set.
+AddRecordFormatString = 'III%ds' % (pttstruct.IDLEN + 1)
+AddRecord = collections.namedtuple('AddRecord', 'userref ctime ipv4 userid')
+REQ_ADD = 1
+_SERVER_ADDR = '127.0.0.1'
+_SERVER_PORT = 5135
+_DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
+			'db_posts.db')
+
+def serialize(data):
+    return json.dumps(data)
+
+def deserialize(data):
+    return json.loads(data)
+
+def DecodeFileHeader(blob):
+    header = pttstruct.unpack_data(blob, pttstruct.FILEHEADER_FMT)
+    return header
+
+def EncodeFileHeader(header):
+    blob = pttstruct.pack_data(header, pttstruct.FILEHEADER_FMT)
+    return blob
+
+def UnpackAddRecord(blob):
+    def strip_if_string(v):
+	return v.strip(chr(0)) if type(v) == str else v
+    data = struct.unpack(AddRecordFormatString, blob)
+    logging.debug("UnpackAddRecord: %r" % (data,))
+    return AddRecord._make(map(strip_if_string, data))
+
+def UnpackPostKey(blob):
+    def strip_if_string(v):
+	return v.strip(chr(0)) if type(v) == str else v
+    data = struct.unpack(PostKeyFormatString, blob)
+    logging.debug("UnpackPostKey: %r" % (data,))
+    return PostKey._make(map(strip_if_string, data))
+
+def PackPost(comment):
+    return struct.pack(PostFormatString, *comment)
+
+def LoadPost(query):
+    logging.debug("LoadPost: %r", query)
+    key = '%s/%s' % (query.board, query.file)
+    num = int(g_db.get(key) or '0')
+    if query.start >= num:
+	return None
+    key += '#%08d' % (query.start + 1)
+    data = g_db.get(key)
+    logging.debug(" => %r", UnpackPost(data))
+    return UnpackPost(data)
+
+
+def SavePost(keypak, data, extra=None):
+    if extra:
+	data.update(extra._asdict())
+    logging.debug("SavePost: %r => %r", keypak, data)
+    key = '%s/%s' % (keypak.board, keypak.file)
+    g_db.set(key, serialize(data))
+    logging.debug(' Saved: %s', key)
+
+def open_database(db_path):
+    global g_db
+
+    class LevelDBWrapper(object):
+	def __init__(self, db):
+	    self.db = db
+
+	def get(self, key):
+	    try:
+		return self.db.Get(key)
+	    except KeyError:
+		return None
+
+	def set(self, key, value):
+	    self.db.Put(key, value)
+
+	def RangeIter(self, **args):
+	    return self.db.RangeIter(*args)
+
+    g_db = LevelDBWrapper(leveldb.LevelDB(db_path))
+    return g_db
+
+def handle_request(socket, _):
+    fd = socket.makefile('rw')
+    fmt_len = '@i'
+    try:
+        req = fd.read(struct.calcsize(RequestFormatString))
+	req = Request._make(struct.unpack(RequestFormatString, req))
+	logging.debug('Found request: %d' % req.operation)
+        # TODO check req.cb
+	if req.operation == REQ_ADD:
+	    header_blob = fd.read(pttstruct.FILEHEADER_SIZE)
+	    addblob = fd.read(struct.calcsize(AddRecordFormatString))
+	    keyblob = fd.read(struct.calcsize(PostKeyFormatString))
+	    SavePost(UnpackPostKey(keyblob), DecodeFileHeader(header_blob),
+		     UnpackAddRecord(addblob))
+	else:
+	    raise ValueError('Unkown operation %d' % req.operation)
+    except:
+        logging.exception("handle_request")
+    finally:
+        try:
+            fd.close()
+            socket.close()
+        except:
+            pass
+
+def main(myname, argv):
+    level = logging.INFO
+    level = logging.WARNING
+    level = logging.DEBUG
+    logging.basicConfig(level=level, format='%(asctime)-15s %(message)s')
+    if len(argv) not in [0, 1]:
+        print "Usage: %s [db_path]" % myname
+        exit(1)
+    db_path = argv[0] if len(argv) > 0 else _DB_PATH
+    logging.warn("Serving at %s:%s [db:%s][pid:%d]...",
+                 _SERVER_ADDR, _SERVER_PORT, db_path, os.getpid())
+    open_database(db_path)
+    server = gevent.server.StreamServer(
+	    (_SERVER_ADDR, _SERVER_PORT), handle_request)
+    server.serve_forever()
+
+if __name__ == '__main__':
+    main(sys.argv[0], sys.argv[1:])
diff --git a/pttbbs/daemon/postd/pyutil b/pttbbs/daemon/postd/pyutil
new file mode 120000
index 00000000..f0598824
--- /dev/null
+++ b/pttbbs/daemon/postd/pyutil
@@ -0,0 +1 @@
+../../util/pyutil
\ No newline at end of file
diff --git a/pttbbs/include/daemons.h b/pttbbs/include/daemons.h
index 2ba227e3..c78c9329 100644
--- a/pttbbs/include/daemons.h
+++ b/pttbbs/include/daemons.h
@@ -178,6 +178,37 @@ typedef struct {
 } PACKSTRUCT CommentQueryRequest;
 
 ///////////////////////////////////////////////////////////////////////
+// Posts Daemon
+//
+#ifndef POSTD_ADDR
+#define POSTD_ADDR   ":5135"
+#endif
+
+enum {
+    POSTD_REQ_ADD = 1,
+};
+
+typedef struct {
+    uint32_t userref; /* user.ctime */
+    uint32_t ctime;   /* post ctime */
+    uint32_t ipv4;    /* user.fromhost */
+    char userid[IDLEN + 1];
+} PACKSTRUCT PostAddExtraInfoReq;
+
+typedef struct {
+    char board[IDLEN + 1];
+    char file[FNLEN + 1];
+} PACKSTRUCT PostKeyReq;
+
+typedef struct {
+    short cb;
+    short operation;
+    fileheader_t header;
+    PostAddExtraInfoReq extra;
+    PostKeyReq key;
+} PACKSTRUCT PostAddRequest;
+
+///////////////////////////////////////////////////////////////////////
 // online friend relation daemon
 //
 typedef struct {
diff --git a/pttbbs/mbbsd/bbs.c b/pttbbs/mbbsd/bbs.c
index 3be8da22..c5f12ed9 100644
--- a/pttbbs/mbbsd/bbs.c
+++ b/pttbbs/mbbsd/bbs.c
@@ -1280,6 +1280,35 @@ does_board_have_public_bm(const boardheader_t *bp) {
     return bp->BM[0] > ' ';
 }
 
+#ifdef USE_POSTD
+static int PostAddRecord(const char *board, const fileheader_t *fhdr,
+                         time4_t ctime)
+{
+    int s;
+    PostAddRequest req = {0};
+
+    req.cb = sizeof(req);
+    req.operation = POSTD_REQ_ADD;
+    strlcpy(req.key.board, board, sizeof(req.key.board));
+    strlcpy(req.key.file, fhdr->filename, sizeof(req.key.file));
+    memcpy(&req.header, fhdr, sizeof(req.header));
+    req.extra.userref = cuser.firstlogin;
+    req.extra.ctime = ctime;
+    req.extra.ipv4 = inet_addr(fromhost);
+    strlcpy(req.extra.userid, cuser.userid, sizeof(req.extra.userid));
+
+    s = toconnectex(POSTD_ADDR, 10);
+    if (s < 0)
+        return 1;
+    if (towrite(s, &req, sizeof(req)) < 0) {
+        close(s);
+        return 1;
+    }
+    close(s);
+    return 0;
+}
+#endif
+
 static int
 do_post_article(int edflags)
 {
@@ -1604,15 +1633,16 @@ do_post_article(int edflags)
 	    // now, genbuf[0] = "if user exists".
 	    if (genbuf[0])
 	    {
-		stampfile(genbuf, &postfile);
+                fileheader_t mailfile;
+                stampfile(genbuf, &mailfile);
 		unlink(genbuf);
 		Copy(fpath, genbuf);
 
-		strlcpy(postfile.owner, cuser.userid, sizeof(postfile.owner));
-		strlcpy(postfile.title, save_title, sizeof(postfile.title));
+		strlcpy(mailfile.owner, cuser.userid, sizeof(mailfile.owner));
+		strlcpy(mailfile.title, save_title, sizeof(mailfile.title));
 		sethomedir(genbuf, quote_user);
                 msg = "回應至作者信箱";
-		if (append_record(genbuf, &postfile, sizeof(postfile)) == -1)
+		if (append_record(genbuf, &mailfile, sizeof(mailfile)) == -1)
 		    msg = err_uid;
 		else
 		    sendalert(quote_user, ALERT_NEW_MAIL);
@@ -1643,11 +1673,11 @@ do_post_article(int edflags)
 	}
 	add_posttimes(usernum, 1);
 #endif
-	// Notify all logins
-	if (addPost)
-	{
-
-	}
+#ifdef USE_POSTD
+        if (edflags & EDITFLAG_KIND_NEWPOST) {
+            PostAddRecord(currboard, &postfile, dashc(fpath));
+        }
+#endif
     }
     pressanykey();
     return FULLUPDATE;