crowbar.swoole.php and crowbar_server.swoole.php

crowbar.swoole.php_and_crowbar_server.swoole.php

========================crowbar.swoole.php:==================
<?php
/*
crowbar.php是客户端
crowbar_server.php是服务端
客户端监听端口 和 服务端转发目标端口 都是写死的
服务端 可以用swoole内建一个socks5, 仿照: http://m.jb51.net/article/83531.htm
然后可以在客户端加参数 来区分forward 和 socks5
*/

$CLOUDFRONT_ID='d11g2fx335xhfk';
$CLIENT_ID=mt_rand();
$HTTP_CONN_POOL=array();
$TCP_CONN_SEQ=array();

$ipsock=new swoole_server('0.0.0.0', 8085, SWOOLE_BASE, SWOOLE_SOCK_TCP);
$ipsock->on('receive', function ($ipsock, $conn, $from_id, $data) use($CLIENT_ID, $CLOUDFRONT_ID, &$HTTP_CONN_POOL, &$TCP_CONN_SEQ)
                       {
                           if (!isset($TCP_CONN_SEQ[$conn]))
                           {echo 'set TCP_CONN_SEQ conn current_seq: '.$conn."\n";
                               $TCP_CONN_SEQ[$conn]['current_seq']=0;
                               $TCP_CONN_SEQ[$conn]['data']=array();
                           }
                           send($CLOUDFRONT_ID, $CLIENT_ID, $ipsock, $conn, $data, $HTTP_CONN_POOL);
                       }
           );
$ipsock->on('close', function ($ipsock, $conn) use(&$TCP_CONN_SEQ)
                     {
                         echo '==============unset conn: '.$conn."\n";
                         //unset($TCP_CONN_SEQ[$conn]);
                     }
           );
$ipsock->on('workerstart', function($ipsock, $worker_id) use($CLOUDFRONT_ID, $CLIENT_ID, &$HTTP_CONN_POOL, &$TCP_CONN_SEQ)
                           {
                               if ($worker_id==0)
                               {
                                   $ipsock->tick(200, function() use($CLOUDFRONT_ID, $CLIENT_ID, &$ipsock, &$HTTP_CONN_POOL, &$TCP_CONN_SEQ)
                                                     {
                                                         receive($CLOUDFRONT_ID, $CLIENT_ID, $ipsock, $HTTP_CONN_POOL, $TCP_CONN_SEQ);
                                                     }
                                                );
                               }
                           }
           );
$ipsock->set(array('worker_num'=>1));      //好像全局变量不能跨worker, 只能设成1个worker
$ipsock->start();

function send($CLOUDFRONT_ID, $CLIENT_ID, $ipsock, $conn, $data, &$http_conn_pool)
{
    //var_export(array_keys($http_conn_pool));
    foreach ($http_conn_pool as $k=>$v)
    {
        if (isset($http_conn_pool[$k]) && $v->isConnected())
        {
            unset($http_conn_pool[$k]);
            $ch=$v;
            break;
        }
        unset($http_conn_pool[$k]);
    }

    if (!isset($ch))
    {
        $ch=new swoole_http_client('52.84.205.98', 443, true);
        $ch->id=''.mt_rand();
        echo "created new client ".($ch->id)."\n";
    }
    $ch->set(['ssl_host_name'=>'a0.awsstatic.com', 'timeout'=>5, 'ssl_verify_peer'=>true, 'keep_alive'=>true]);
    $ch->setHeaders(array('Host'=>$CLOUDFRONT_ID.'.cloudfront.net'));
    echo 'to post:'.microtime(1).':'.strlen($data)."\n";
    $ch->post('/post?client='.$CLIENT_ID.'&conn='.$conn,
              $data,
              function($ch) use (&$ipsock, $conn, &$http_conn_pool)
              {
                  //放到连接池
                  if ($ch->isConnected())
                  {
                      $http_conn_pool[$ch->id]=$ch;
                  }

                  if ($ch->body!=='ok')
                  {
                      echo "===================conn ".$conn." closed!";
                      $ipsock->close($conn);
                  }
              }
             );
}

function receive($CLOUDFRONT_ID, $CLIENT_ID, $ipsock, &$http_conn_pool, &$TCP_CONN_SEQ)
{
    //var_export(array_keys($http_conn_pool));
    foreach ($http_conn_pool as $k=>$v)
    {
        if (isset($http_conn_pool[$k]) && $v->isConnected())
        {
            unset($http_conn_pool[$k]);
            $ch=$v;
            break;
        }
        unset($http_conn_pool[$k]);
    }

    if (!isset($ch))
    {
        $ch=new swoole_http_client('52.84.205.98', 443, true);
        $ch->id=''.mt_rand();
        echo "created new client ".($ch->id)."\n";
    }
    $ch->set(['ssl_host_name'=>'a0.awsstatic.com', 'timeout'=>10, 'ssl_verify_peer'=>true, 'keep_alive'=>true]);
    $ch->setHeaders(array('Host'=>$CLOUDFRONT_ID.'.cloudfront.net'));
    $ch->post('/get?client='.$CLIENT_ID, array(''=>''), function ($ch) use(&$ipsock, &$http_conn_pool, &$TCP_CONN_SEQ)
                  {
                      //放到连接池
                      if ($ch->isConnected())
                      {
                          $http_conn_pool[$ch->id]=$ch;
                      }

                      $datas=@json_decode($ch->body, 1);
                      if (is_array($datas))
                      {
                          foreach ($datas as $conn=>$data_seqs)
                          {echo 'receive conn: '.$conn."\n";var_export(array_keys($TCP_CONN_SEQ));
                              foreach ($data_seqs as $seq=>$data_seq)
                              {
                                  $TCP_CONN_SEQ[$conn]['data'][$seq]=base64_decode($data_seq['base64_data']);
                              }
                              ksort($TCP_CONN_SEQ[$conn]['data']);
                              foreach ($TCP_CONN_SEQ[$conn]['data'] as $seq=>$data_seq)
                              {
                                  if ($seq===$TCP_CONN_SEQ[$conn]['current_seq'])
                                  {
                                      $ipsock->send($conn, $data_seq);
                                      $TCP_CONN_SEQ[$conn]['current_seq']=$TCP_CONN_SEQ[$conn]['current_seq']+1;
                                      echo 'received:'.microtime(1).':'.strlen($data_seq)."\n";
                                  }
                              }
                          }
                      }
                  }
            );
}
?>

=========================aws cloud front:=================================
....cloudfront.net在aws里配置了映射到:
Origin Domain Name:<vps域名>
Origin Protocol Policy:HTTP Only
HTTP Port:8080
Viewer Protocol Policy:HTTPS Only
Allowed HTTP Methods:GET, HEAD, OPTIONS, PUT, POST, PATCH, DELETE
Object Caching:Customize
Maximum TTL:1
Default TTL:1
Query String Forwarding and Caching:Forward all,cache based on all
Price Class:Use Only US, Canada and Europe
SSL Certificate:Default CloudFront Certificate (*.cloudfront.net)
其它都默认即可

========================crowbar_server.swoole.php:==================
<?php
$CONNS=array();
$RESPONSES=array();
$SOCKS=explode(':', '127.0.0.1:8081');

$server=new swoole_http_server('0.0.0.0', 8080);
$server->on('request', function($req, $res) use(&$CONNS, &$RESPONSES, $SOCKS)
                       {
                           if (!$req->get['client'])
                           {
                               $res->end('invalid');
                               return;
                           }
                           if ($req->server['request_uri']==='/post')
                           {
                               $req_data=$req->rawContent();
                               if (!isset($CONNS[$req->get['client'].':'.$req->get['conn']]))
                               {
                                   $cli=new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
                                   $cli->client=$req->get['client'];
                                   $cli->conn=$req->get['conn'];
                                   $cli->on('connect', function($cli) use(&$CONNS, &$req, &$res, $req_data)
                                                       {
                                                           $CONNS[$req->get['client'].':'.$req->get['conn']]=$cli;
                                                           $CONNS[$req->get['client'].':'.$req->get['conn']]->send($req_data);
                                                           echo microtime(1).', conn-req: client-'.$req->get['client'].',conn-'.$req->get['conn'].',length:'.strlen($req_data)."\n";
                                                           $res->end('ok');
                                                       }
                                           );
                                   $cli->on('error', function($cli) {echo 'error';});
                                   $cli->seq=-1;
                                   $cli->on('receive', function($cli, $data) use(&$RESPONSES)
                                                          {
                                                              $cli->seq=$cli->seq+1;
                                                              $RESPONSES[$cli->client][$cli->conn][$cli->seq]=array('ori_len'=>strlen($data), 'base64_data'=>base64_encode($data));
                                                          }
                                              );
                                   $cli->on('close', function($cli) {unset($CONNS[$req->get['client'].':'.$req->get['conn']]);});
                                   $cli->connect($SOCKS[0], $SOCKS[1]);
                               }
                               else
                               {
                                   $CONNS[$req->get['client'].':'.$req->get['conn']]->send($req_data);
                                   echo microtime(1).', req: client-'.$req->get['client'].',conn-'.$req->get['conn'].',length:'.strlen($req_data)."\n";
                                   $res->end('ok');
                               }
                           }
                           else
                           {
                               echo 'clients count: '.count($RESPONSES)."\n";
                               $tmp=$RESPONSES[$req->get['client']];
                               if (!$tmp)
                               {
                                   @$res->end('');
                                   return;
                               }
                               unset($RESPONSES[$req->get['client']]);
                               foreach ($tmp as $k=>$v)
                               {
                                   foreach ($v as $d)
                                   {
                                       echo microtime(1).', resp: client-'.$req->get['client'].',conn-'.$k.',lenth:'.$d['ori_len'].'|';
                                   }
                               }
                               echo "\n";
                               $res->end(json_encode($tmp));
                           }
                       }
           );
$server->set(array('worker_num'=>1,               //好像全局变量不能跨worker, 只能设成1个worker
                   'http_parse_post'=>false
                  )
            );
$server->start();
?>