a48aa1fe34740ebc6c222ef827ecbbc56448ed64
[brisk.git] / web / spush / brisk-spush.php
1 #!/usr/bin/php
2 <?php
3 /*
4  *  brisk - spush/brisk-spush.php
5  *
6  *  Copyright (C) 2012 Matteo Nastasi
7  *                          mailto: nastasi@alternativeoutput.it 
8  *                                  matteo.nastasi@milug.org
9  *                          web: http://www.alternativeoutput.it
10  *
11  * This program is free software; you can redistribute it and/or modify
12  * it under the terms of the GNU General Public License as published by
13  * the Free Software Foundation; either version 2 of the License, or
14  * (at your option) any later version.
15  *
16  * This program is distributed in the hope that it will be useful, but
17  * WITHOUT ANY WARRANTY; without even the implied warranty of
18  * MERCHANTABLILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
19  * General Public License for more details. You should have received a
20  * copy of the GNU General Public License along with this program; if
21  * not, write to the Free Software Foundation, Inc, 59 Temple Place -
22  * Suite 330, Boston, MA 02111-1307, USA.
23  *
24  * TODO
25  *
26  *
27  *   - BUG: logout failed
28  *   - BUG: fast loop on stream index_rd_ifra page
29  *
30  *   - garbage management
31  *   - log_legal address fix
32  *   - from room to table
33  *   - from table to room
34  *   - index_wr other issues
35  *   - manage and test cross forwarder between table and room
36  *   - setcookie (for tables only)
37  *   - keepalive management
38  *
39  *   DONE/FROZEN - problema con getpeer (HOSTADDR)
40  *
41  *   DONE - chunked
42  *   DONE - bug: read from a not resource handle (already closed because a new socket substitute it)
43  *   DONE - partial write for normal page management
44  *   DONE - index_rd_ifra: last_clean issue
45  *   DONE - fwrite failed error management (select, buffer where store unsent data, and fwrite check and retry)
46  *   ABRT - index_wr.php::reload - reload is js-only function
47  *   DONE - bug: after restart index_rd.php receive from prev clients a lot of req
48  *   DONE - index_wr.php::chat
49  *   DONE - index_wr.php::exit
50  *   DONE - index_rd.php porting
51  *   DONE - generic var management from internet
52  *   DONE - index.php auth part
53  */
54
55 $G_base = "../";
56
57 require_once("./sac-a-push.phh");
58 require_once("./brisk-spush.phh");
59 require_once($G_base."Obj/user.phh");
60 require_once($G_base."Obj/brisk.phh");
61 require_once($G_base."Obj/auth.phh");
62 // require_once("../Obj/proxyscan.phh");
63 require_once($G_base."index.php");
64 require_once($G_base."index_wr.php");
65 require_once($G_base."index_rd_ifra.php");
66 require_once($G_base."briskin5/Obj/briskin5.phh");
67
68 define('SITE_PREFIX', '/brisk/');
69
70 function headers_render($header, $len)
71 {
72     
73     $s = "";
74     $s .= "HTTP/1.1 200 OK\r\n";
75     if (!isset($header['Date']))
76         $s .= sprintf("Date: %s\r\n", date(DATE_RFC822));
77     if (!isset($header['Connection']))
78         $s .= "Connection: close\r\n";
79     if (!isset($header['Content-Type']))
80         $s .= "Content-Type: text/html\r\n";
81     foreach($header as $key => $value) {
82         $s .= sprintf("%s: %s\r\n", $key, $value);
83     }
84     if ($len == -1) {
85         $s .= "Cache-Control: no-cache, must-revalidate\r\n";
86         $s .= "Expires: Mon, 26 Jul 1997 05:00:00 GMT\r\n";
87         $s .= "Content-Encoding: chunked\r\n";
88         $s .= "Transfer-Encoding: chunked\r\n";
89     }
90     else if ($len > 0) {
91         $s .= sprintf("Content-Length: %d\r\n", $len);
92     }
93     $s .= "\r\n";
94
95     return ($s);
96 }
97
98 /*
99  *  Caching system using ob php system to cache old style pages
100  *  to a var and than send it with more calm
101  */
102
103 function shutta()
104 {
105   log_rd2("SHUTTA [".connection_status()."] !");
106 }
107
108 register_shutdown_function('shutta');
109
110 /*
111  *  MAIN
112  */
113
114 function chunked_content($content)
115 {
116     $content_l = mb_strlen($content, "ASCII");
117
118     return (sprintf("%X\r\n%s\r\n", $content_l, $content));
119 }
120
121 function chunked_fini()
122 {
123     return sprintf("0\r\n");
124 }
125
126 class Sac_a_push {
127     static $fixed_fd = 2;
128     
129     var $file_socket;
130     var $unix_socket;
131     var $socks;
132     var $s2u;
133     var $pages_flush;
134
135     var $list;
136     var $in;
137
138     var $debug;
139     var $blocking_mode;
140
141     var $room;
142     var $bin5;
143
144     var $curtime;
145
146     var $rndstr;
147     var $main_loop;
148
149     function Sac_a_push()
150     {
151     }
152
153     // Sac_a_push::create("/tmp/brisk.sock", 0, 0
154
155     static function create(&$room, $sockname, $debug, $blocking_mode)
156     {        
157         $thiz = new Sac_a_push();
158         
159         $thiz->room = $room;
160         $thiz->file_socket = $sockname;
161         $thiz->unix_socket = "unix://$sockname";
162         $thiz->debug = $debug;
163         $thiz->socks = array();
164         $thiz->s2u  = array();
165         $thiz->pages_flush = array();
166
167         $thiz->blocking_mode = 0; // 0 for non-blocking
168
169         $thiz->rndstr = "";
170         for ($i = 0 ; $i < 4096 ; $i++) {
171             $thiz->rndstr .= chr(mt_rand(65, 90));
172         }
173         
174         if (file_exists($thiz->file_socket)) {
175             unlink($thiz->file_socket);
176         }
177     
178         $old_umask = umask(0);
179         if (($thiz->list = stream_socket_server($thiz->unix_socket, $err, $errs)) === FALSE) {
180             return (FALSE);
181         }
182         umask($old_umask);
183         stream_set_blocking($thiz->list, $thiz->blocking_mode); # Set the stream to non-blocking
184
185         if (($thiz->in = fopen("php://stdin", "r")) === FALSE) {
186             return(FALSE);
187         }
188
189         $thiz->main_loop = FALSE;
190
191         return ($thiz);
192     }
193
194     function socks_set($sock, $user)
195     {
196         $id = intval($sock);
197
198         $this->s2u[$id]   = $user;
199         $this->socks[$id] = $sock;
200     }
201
202     function socks_unset($sock)
203     {
204         $id = intval($sock);
205
206         unset($this->s2u[$id]);
207         unset($this->socks[$id]);
208     }
209
210     function pgflush_try_add(&$new_socket, $tout, $header_out, $content)
211     {
212         $pgflush = new PageFlush($new_socket, $this->curtime, $tout, $header_out, $content);
213
214         if ($pgflush->try_flush($this->curtime) == FALSE) {
215             // Add $pgflush to the pgflush array
216             $this->pgflush_add($pgflush);
217         }
218     }
219
220     function pgflush_add($pgflush)
221     {
222         array_push($this->pages_flush, $pgflush);
223     }
224
225     function run()
226     {
227         if ($this->main_loop) {
228             return (FALSE);
229         }
230         
231         $this->main_loop = TRUE;
232         
233         while ($this->main_loop) {
234             $this->curtime = time();
235             printf("IN LOOP: Current opened: %d  pages_flush: %d - ", count($this->socks), count($this->pages_flush));
236             
237             /* Prepare the read array */
238             /* // when we manage it ... */
239             /* if ($shutdown)  */
240             /*     $read   = array_merge(array("$in" => $in), $socks); */
241             /* else */
242             $read   = array_merge(array(intval($this->list) => $this->list, intval($this->in) => $this->in),
243                                   $this->socks);
244             
245             if ($this->debug > 1) {
246                 printf("PRE_SELECT\n");
247                 print_r($read);
248             }
249             $write  = NULL;
250             $except = NULL;
251             $num_changed_sockets = stream_select($read, $write, $except, 0, 250000);
252         
253             if ($num_changed_sockets == 0) {
254                 printf(" no data in 5 secs ");
255             } 
256             else if ($num_changed_sockets > 0) {
257                 printf("num sock %d num_of_socket: %d\n", $num_changed_sockets, count($read));
258                 if ($this->debug > 1) {
259                     print_r($read);
260                 }
261                 /* At least at one of the sockets something interesting happened */
262                 foreach ($read as $i => $sock) {
263                     /* is_resource check is required because there is the possibility that
264                        during new request an old connection is closed */
265                     if (!is_resource($sock)) {
266                         continue;
267                     }
268                     if ($sock === $this->list) {
269                         printf("NUOVA CONNEX\n");
270                         $new_unix = stream_socket_accept($this->list);
271                         $stream_info = "";
272                         $method      = "";
273                         $get         = array();
274                         $post        = array();
275                         $cookie      = array();
276                         if (($new_socket = ancillary_getstream($new_unix, $stream_info)) !== FALSE) {
277                             printf("NEW_SOCKET: %d\n", intval($new_socket));
278                             stream_set_blocking($new_socket, $this->blocking_mode); // Set the stream to non-blocking
279                             printf("RECEIVED HEADER:\n%s", $stream_info);
280                             $path = spu_process_info($stream_info, $method, $header, $get, $post, $cookie);
281                             printf("PATH: [%s]\n", $path);
282                             printf("M: %s\nHEADER:\n", $method);
283                             print_r($header);
284                             printf("GET:\n");
285                             print_r($get);
286                             printf("POST:\n");
287                             print_r($post);
288                             printf("COOKIE:\n");
289                             print_r($cookie);
290
291                             $addr = stream_socket_get_name($new_socket, TRUE);
292                             $header_out = array();
293
294                             $this->room->request_mgr($this, $header_out, $new_socket, $path, $addr, $get, $post, $cookie);
295                             printf("number of sockets after %d\n", count($this->socks));
296                         }
297                         else {
298                             printf("WARNING: ancillary_getstream failed\n");
299                         }
300                     }
301                     else {
302                         if (($buf = fread($sock, 512)) === FALSE) {
303                             printf("error read\n");
304                             exit(123);
305                         }
306                         else if (strlen($buf) === 0) {
307                             if ($sock === $this->list) {
308                                 printf("Arrivati %d bytes da list\n", strlen($buf));
309                             }
310                             else if ($sock === $this->in) {
311                                 printf("Arrivati %d bytes da stdin\n", strlen($buf));
312                             }
313                             else {
314                                 // $user_a[$s2u[intval($sock)]]->disable();
315                                 if ($this->s2u[intval($sock)]->rd_socket_get() != NULL) {
316                                     $this->s2u[intval($sock)]->rd_socket_set(NULL);
317                                 }
318                                 unset($this->socks[intval($sock)]);
319                                 unset($this->s2u[intval($sock)]);
320                                 fclose($sock);
321                                 printf("CLOSE ON READ\n");
322                             }
323                             if ($this->debug > 1) {
324                                 printf("post unset\n");
325                                 print_r($this->socks);
326                             }
327                         }
328                         else {
329                             if ($debug > 1) {
330                                 print_r($read);
331                             }
332                             if ($sock === $this->list) {
333                                 printf("Arrivati %d bytes da list\n", strlen($buf));
334                             }
335                             else if ($sock === $this->in) {
336                                 printf("Arrivati %d bytes da stdin\n", strlen($buf));
337                             }
338                             else {
339                                 $key = array_search("$sock", $this->socks);
340                                 printf("Arrivati %d bytes dalla socket n. %d\n", strlen($buf), $key);
341                             }
342                         }
343                     }
344                 }
345             }
346             
347
348             /* manage unfinished pages */
349             foreach ($this->pages_flush as $k => $pgflush) {
350                 if ($pgflush->try_flush($this->curtime) == TRUE) {
351                     unset($this->pages_flush[$k]);
352                 }
353             }
354             
355             /* manage open streaming */
356             foreach ($this->socks as $k => $sock) {
357                 if (isset($this->s2u[intval($sock)])) {
358                     $user = $this->s2u[intval($sock)];
359                     $response = $user->rd_cache_get();
360                     if ($response == "") {
361                         $content = "";
362                         $user->stream_main($content, $get, $post, $cookie);
363                         
364                         if ($content == "" && $user->rd_kalive_is_expired($this->curtime)) {
365                             $content = $user->stream_keepalive();
366                         }
367                         if ($content != "") {
368                             $response = chunked_content($content);
369                         }
370                     }
371                     
372                     if ($response != "") {
373                         echo "SPIA: [".substr($response, 0, 60)."...]\n";
374                         $response_l = mb_strlen($response, "ASCII");
375                         $wret = @fwrite($sock, $response);
376                         if ($wret < $response_l) {
377                             printf("TROUBLE WITH FWRITE: %d\n", $wret);
378                             $user->rd_cache_set(mb_substr($response, $wret, $response_l - $wret, "ASCII"));
379                         }
380                         else {
381                             $user->rd_cache_set("");
382                         }
383                         fflush($sock);
384                         $user->rd_kalive_reset($this->curtime);
385                     }
386                     
387                     // close socket after a while to prevent client memory consumption
388                     if ($user->rd_endtime_is_expired($this->curtime)) {
389                         if ($this->s2u[intval($sock)]->rd_socket_get() != NULL) {
390                             $this->s2u[intval($sock)]->rd_socket_set(NULL);
391                         }
392                         unset($this->socks[intval($sock)]);
393                         unset($this->s2u[intval($sock)]);
394                         fclose($sock);
395                         printf("CLOSE ON LOOP\n");
396                     }
397                 }
398             }  // foreach ($this->socks...
399             printf("\n");
400         }  // while (...
401     }  // function run(...
402 }
403
404 function main()
405 {
406     if (($room = Room::create()) == FALSE) {
407         log_crit("room::create failed");
408         exit(1);
409     }
410
411     if (($s_a_p = Sac_a_push::create($room, "/tmp/brisk.sock", 0, 0)) === FALSE) {
412         exit(1);
413     }
414
415     $s_a_p->run();
416
417     exit(0);
418 }
419
420 main();
421 ?>