ae.net.nntp.listener source code

1 /**
2  * NNTP listener (periodically poll server for new messages).
3  *
4  * License:
5  *   This Source Code Form is subject to the terms of
6  *   the Mozilla Public License, v. 2.0. If a copy of
7  *   the MPL was not distributed with this file, You
8  *   can obtain one at http://mozilla.org/MPL/2.0/.
9  *
10  * Authors:
11  *   Vladimir Panteleev <vladimir@thecybershadow.net>
12  */
13 
14 module ae.net.nntp.listener;
15 
16 import ae.net.nntp.client;
17 
18 import std.datetime;
19 import std.typecons;
20 
21 import ae.sys.timing;
22 import ae.sys.log;
23 
24 const POLL_PERIOD = 2.seconds;
25 
26 class NntpListener
27 {
28 private:
29 	typeof(scoped!NntpClient(Logger.init)) client;
30 	string server;
31 	string lastDate;
32 	bool[string] oldMessages;
33 	TimerTask pollTimer;
34 	bool connected, polling;
35 	int queued;
36 
37 	void reconnect()
38 	{
39 		assert(!connected);
40 		client.connect(server, &onConnect);
41 	}
42 
43 	void schedulePoll()
44 	{
45 		pollTimer = setTimeout(&poll, POLL_PERIOD);
46 	}
47 
48 	void poll()
49 	{
50 		pollTimer = null;
51 		client.getDate(&onDate);
52 		client.getNewNews("*", lastDate[0..8] ~ " " ~ lastDate[8..14] ~ " GMT", &onNewNews);
53 	}
54 
55 	void onConnect()
56 	{
57 		connected = true;
58 		queued = 0;
59 
60 		if (polling)
61 		{
62 			if (lastDate)
63 				poll();
64 			else
65 				client.getDate(&onDate);
66 		}
67 	}
68 
69 	void onDisconnect(string reason, DisconnectType type)
70 	{
71 		connected = false;
72 		if (polling)
73 		{
74 			if (pollTimer && pollTimer.isWaiting())
75 				pollTimer.cancel();
76 			if (type != DisconnectType.requested)
77 				setTimeout(&reconnect, 10.seconds);
78 		}
79 	}
80 
81 	void onDate(string date)
82 	{
83 		if (polling)
84 		{
85 			if (lastDate is null)
86 				schedulePoll();
87 			lastDate = date;
88 		}
89 	}
90 
91 	void onNewNews(string[] reply)
92 	{
93 		auto messages = reply[1..$];
94 
95 		assert(queued == 0);
96 		foreach (message; messages)
97 			if (message !in oldMessages)
98 			{
99 				oldMessages[message] = true;
100 				client.getMessage(message, &onMessage);
101 				queued++;
102 			}
103 		if (queued==0)
104 			schedulePoll();
105 	}
106 
107 	void onMessage(string[] lines, string num, string id)
108 	{
109 		if (handleMessage)
110 			handleMessage(lines, num, id);
111 
112 		if (polling)
113 		{
114 			queued--;
115 			if (queued==0)
116 				schedulePoll();
117 		}
118 	}
119 
120 public:
121 	this(Logger log)
122 	{
123 		client = scoped!NntpClient(log);
124 		client.handleDisconnect = &onDisconnect;
125 	}
126 
127 	void connect(string server)
128 	{
129 		this.server = server;
130 		reconnect();
131 	}
132 
133 	void disconnect()
134 	{
135 		client.disconnect();
136 	}
137 
138 	void startPolling(string lastDate = null)
139 	{
140 		assert(!polling, "Already polling");
141 		polling = true;
142 		this.lastDate = lastDate;
143 		if (connected)
144 			poll();
145 	}
146 
147 	void delegate(string[] lines, string num, string id) handleMessage;
148 }