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 $rndstr;
 145     var $main_loop;
 146
 147     function Sac_a_push()
 148     {
 149     }
 150
 151     // Sac_a_push::create("/tmp/brisk.sock", 0, 0
 152
 153     static function create($sockname, $debug, $blocking_mode)
 154     {
 155         $thiz = new Sac_a_push();
 156
 157         $thiz->file_socket = $sockname;
 158         $thiz->unix_socket = "unix://$sockname";
 159         $thiz->debug = $debug;
 160         $thiz->socks = array();
 161         $thiz->s2u  = array();
 162         $thiz->pages_flush = array();
 163
 164         $thiz->blocking_mode = 0; // 0 for non-blocking
 165
 166         if (($thiz->room = Room::create()) == FALSE) {
 167             log_crit("room::create failed");
 168             return FALSE;
 169         }
 170
 171
 172         $thiz->rndstr = "";
 173         for ($i = 0 ; $i < 4096 ; $i++) {
 174             $thiz->rndstr .= chr(mt_rand(65, 90));
 175         }
 176
 177         if (file_exists($thiz->file_socket)) {
 178             unlink($thiz->file_socket);
 179         }
 180
 181         $old_umask = umask(0);
 182         if (($thiz->list = stream_socket_server($thiz->unix_socket, $err, $errs)) === FALSE) {
 183             return (FALSE);
 184         }
 185         umask($old_umask);
 186         stream_set_blocking($thiz->list, $thiz->blocking_mode); # Set the stream to non-blocking
 187
 188         if (($thiz->in = fopen("php://stdin", "r")) === FALSE) {
 189             return(FALSE);
 190         }
 191
 192         $thiz->main_loop = FALSE;
 193
 194         return ($thiz);
 195     }
 196
 197     function run()
 198     {
 199         if ($this->main_loop) {
 200             return (FALSE);
 201         }
 202
 203         $this->main_loop = TRUE;
 204
 205         while ($this->main_loop) {
 206             $curtime = time();
 207             printf("IN LOOP: Current opened: %d  pages_flush: %d\n", count($this->socks), count($this->pages_flush));
 208
 209             /* Prepare the read array */
 210             /* // when we manage it ... */
 211             /* if ($shutdown)  */
 212             /*     $read   = array_merge(array("$in" => $in), $socks); */
 213             /* else */
 214             $read   = array_merge(array(intval($this->list) => $this->list, intval($this->in) => $this->in),
 215                                   $this->socks);
 216
 217             if ($this->debug > 1) {
 218                 printf("PRE_SELECT\n");
 219                 print_r($read);
 220             }
 221             $write  = NULL;
 222             $except = NULL;
 223             $num_changed_sockets = stream_select($read, $write, $except, 0, 250000);
 224
 225             if ($num_changed_sockets == 0) {
 226                 printf("No data in 5 secs\n");
 227             }
 228             else if ($num_changed_sockets > 0) {
 229                 printf("num sock %d num_of_socket: %d\n", $num_changed_sockets, count($read));
 230                 if ($this->debug > 1) {
 231                     print_r($read);
 232                 }
 233                 /* At least at one of the sockets something interesting happened */
 234                 foreach ($read as $i => $sock) {
 235                     /* is_resource check is required because there is the possibility that
 236                        during new request an old connection is closed */
 237                     if (!is_resource($sock)) {
 238                         continue;
 239                     }
 240                     if ($sock === $this->list) {
 241                         printf("NUOVA CONNEX\n");
 242                         $new_unix = stream_socket_accept($this->list);
 243                         $stream_info = "";
 244                         $method      = "";
 245                         $get         = array();
 246                         $post        = array();
 247                         $cookie      = array();
 248                         if (($new_socket = ancillary_getstream($new_unix, $stream_info)) !== FALSE) {
 249                             stream_set_blocking($new_socket, $this->blocking_mode); // Set the stream to non-blocking
 250                             printf("RECEIVED HEADER:\n%s", $stream_info);
 251                             $path = spu_process_info($stream_info, $method, $header, $get, $post, $cookie);
 252                             printf("PATH: [%s]\n", $path);
 253                             printf("M: %s\nHEADER:\n", $method);
 254                             print_r($header);
 255                             printf("GET:\n");
 256                             print_r($get);
 257                             printf("POST:\n");
 258                             print_r($post);
 259                             printf("COOKIE:\n");
 260                             print_r($cookie);
 261
 262                             $addr = stream_socket_get_name($new_socket, TRUE);
 263                             $header_out = array();
 264
 265                             switch ($path) {
 266                             case SITE_PREFIX:
 267                             case SITE_PREFIX."index.php":
 268                                 ob_start();
 269                                 index_main($this->room, $header_out, $addr, $get, $post, $cookie);
 270                                 $content = ob_get_contents();
 271                                 ob_end_clean();
 272
 273                                 $pgflush = new PageFlush($new_socket, $curtime, 20, $header_out, $content);
 274
 275                                 if ($pgflush->try_flush($curtime) == FALSE) {
 276                                     // Add $pgflush to the pgflush array
 277                                     array_push($this->pages_flush, $pgflush);
 278                                 }
 279
 280                                 break;
 281                             case SITE_PREFIX."index_wr.php":
 282                                 ob_start();
 283                                 index_wr_main($this->room, $addr, $get, $post, $cookie);
 284                                 $content = ob_get_contents();
 285                                 ob_end_clean();
 286
 287                                 $pgflush = new PageFlush($new_socket, $curtime, 20, $header_out, $content);
 288
 289                                 if ($pgflush->try_flush($curtime) == FALSE) {
 290                                     // Add $pgflush to the pgflush array
 291                                     array_push($this->pages_flush, $pgflush);
 292                                 }
 293                             break;
 294                             case SITE_PREFIX."index_rd_ifra.php":
 295                                 do {
 296                                     if (!isset($cookie['sess'])
 297                                         || (($user = $this->room->get_user($cookie['sess'], $idx)) == FALSE)) {
 298                                         $content = User::stream_fini(TRUE);
 299
 300                                         $pgflush = new PageFlush($new_socket, $curtime, 20, $header_out, $content);
 301
 302                                         if ($pgflush->try_flush($curtime) == FALSE) {
 303                                             // Add $pgflush to the pgflush array
 304                                             array_push($this->pages_flush, $pgflush);
 305                                         }
 306                                         break;
 307                                     }
 308                                     // close a previous opened index_read_ifra socket, if exists
 309                                     if (($prev = $user->rd_socket_get()) != NULL) {
 310                                         unset($this->s2u[intval($user->rd_socket_get())]);
 311                                         unset($this->socks[intval($user->rd_socket_get())]);
 312                                         fclose($user->rd_socket_get());
 313                                         printf("CLOSE AND OPEN AGAIN ON IFRA2\n");
 314                                         $user->rd_socket_set(NULL);
 315                                     }
 316
 317                                     $content = "";
 318                                     $user->stream_init($header_out, $content, $get, $post, $cookie);
 319
 320                                     $response = headers_render($header_out, -1).chunked_content($content);
 321                                     $response_l = mb_strlen($response, "ASCII");
 322
 323                                     $wret = @fwrite($new_socket, $response, $response_l);
 324                                     if ($wret < $response_l) {
 325                                         printf("TROUBLES WITH FWRITE: %d\n", $wret);
 326                                         $user->rd_cache_set(mb_substr($content, $wret, $response_l - $wret, "ASCII"));
 327                                     }
 328                                     else {
 329                                         $user->rd_cache_set("");
 330                                     }
 331                                     fflush($new_socket);
 332
 333                                     $this->s2u[intval($new_socket)] = $user;
 334                                     $this->socks[intval($new_socket)] = $new_socket;
 335                                     $user->rd_socket_set($new_socket);
 336                                 } while (FALSE);
 337
 338                                 break;
 339
 340                                 /* default: */
 341                                 /*     $cl = strlen(SITE_PREFIX."briskin5/"); */
 342                                 /*     if (!strncmp($this->path, SITE_PREFIX."briskin5/", $cl)) { */
 343                                 /*         Bin5::page_manager($room, $header_out, substr($path,$cl), $method, $addr, $get, $post, $cookie); */
 344                             }
 345                         }
 346                         else {
 347                             printf("WARNING: ancillary_getstream failed\n");
 348                         }
 349                     }
 350                     else {
 351                         if (($buf = fread($sock, 512)) === FALSE) {
 352                             printf("error read\n");
 353                             exit(123);
 354                         }
 355                         else if (strlen($buf) === 0) {
 356                             if ($sock === $this->list) {
 357                                 printf("Arrivati %d bytes da list\n", strlen($buf));
 358                             }
 359                             else if ($sock === $this->in) {
 360                                 printf("Arrivati %d bytes da stdin\n", strlen($buf));
 361                             }
 362                             else {
 363                                 // $user_a[$s2u[intval($sock)]]->disable();
 364                                 if ($this->s2u[intval($sock)]->rd_socket_get() != NULL) {
 365                                     $this->s2u[intval($sock)]->rd_socket_set(NULL);
 366                                 }
 367                                 unset($this->socks[intval($sock)]);
 368                                 unset($this->s2u[intval($sock)]);
 369                                 fclose($sock);
 370                                 printf("CLOSE ON READ\n");
 371                             }
 372                             if ($this->debug > 1) {
 373                                 printf("post unset\n");
 374                                 print_r($this->socks);
 375                             }
 376                         }
 377                         else {
 378                             if ($debug > 1) {
 379                                 print_r($read);
 380                             }
 381                             if ($sock === $this->list) {
 382                                 printf("Arrivati %d bytes da list\n", strlen($buf));
 383                             }
 384                             else if ($sock === $this->in) {
 385                                 printf("Arrivati %d bytes da stdin\n", strlen($buf));
 386                             }
 387                             else {
 388                                 $key = array_search("$sock", $this->socks);
 389                                 printf("Arrivati %d bytes dalla socket n. %d\n", strlen($buf), $key);
 390                             }
 391                         }
 392                     }
 393                 }
 394             }
 395
 396
 397             /* manage unfinished pages */
 398             foreach ($this->pages_flush as $k => $pgflush) {
 399                 if ($pgflush->try_flush($curtime) == TRUE) {
 400                     unset($this->pages_flush[$k]);
 401                 }
 402             }
 403
 404             /* manage open streaming */
 405             foreach ($this->socks as $k => $sock) {
 406                 if (isset($this->s2u[intval($sock)])) {
 407                     $user = $this->s2u[intval($sock)];
 408                     $response = $user->rd_cache_get();
 409                     if ($response == "") {
 410                         $content = "";
 411                         $user->stream_main($content, $get, $post, $cookie);
 412
 413                         if ($content == "" && $user->rd_kalive_is_expired($curtime)) {
 414                             $content = $user->stream_keepalive();
 415                         }
 416                         if ($content != "") {
 417                             $response = chunked_content($content);
 418                         }
 419                     }
 420
 421                     if ($response != "") {
 422                         echo "SPIA: [".substr($response, 0, 60)."...]\n";
 423                         $response_l = mb_strlen($response, "ASCII");
 424                         $wret = @fwrite($sock, $response);
 425                         if ($wret < $response_l) {
 426                             printf("TROUBLE WITH FWRITE: %d\n", $wret);
 427                             $user->rd_cache_set(mb_substr($response, $wret, $response_l - $wret, "ASCII"));
 428                         }
 429                         else {
 430                             $user->rd_cache_set("");
 431                         }
 432                         fflush($sock);
 433                         $user->rd_kalive_reset($curtime);
 434                     }
 435
 436                     // close socket after a while to prevent client memory consumption
 437                     if ($user->rd_endtime_is_expired($curtime)) {
 438                         // $user_a[$s2u[intval($sock)]]->disable();
 439                         if ($this->s2u[intval($sock)]->rd_socket_get() != NULL) {
 440                             $this->s2u[intval($sock)]->rd_socket_set(NULL);
 441                         }
 442                         unset($this->socks[intval($sock)]);
 443                         unset($this->s2u[intval($sock)]);
 444                         fclose($sock);
 445                         printf("CLOSE ON LOOP\n");
 446                     }
 447                 }
 448             }
 449         }
 450     }
 451 }
 452
 453 function main()
 454 {
 455     if (($s_a_p = Sac_a_push::create("/tmp/brisk.sock", 0, 0)) === FALSE) {
 456         exit(1);
 457     }
 458
 459     $s_a_p->run();
 460
 461     exit(0);
 462 }
 463
 464 main();
 465 ?>