view release on metacpan or  search on metacpan

lib/AnyEvent/SSH2.pm  view on Meta::CPAN

# $Id: SSH2.pm,v 1.47 2009/01/26 01:50:38 turnstep Exp $
package AnyEvent::SSH2;
use strict;
use AE;
use AnyEvent::Handle;
use Net::SSH::Perl::Kex;
use Net::SSH::Perl::ChannelMgr;
use Net::SSH::Perl::Packet;
use Net::SSH::Perl::Buffer;
use Net::SSH::Perl::Constants qw( :protocol :msg2 :compat :hosts :channels :proposal :kex
                                  CHAN_INPUT_CLOSED CHAN_INPUT_WAIT_DRAIN );
use Net::SSH::Perl::Cipher;
use Net::SSH::Perl::AuthMgr;
use Net::SSH::Perl::Comp;
use Net::SSH::Perl::Util qw(:hosts);
use Scalar::Util qw(blessed weaken);
use Carp qw( croak );

use base qw( Net::SSH::Perl );
our $VERSION = '0.04';

lib/AnyEvent/SSH2.pm  view on Meta::CPAN

    $cmgr->process_output_packets;

    # 如果处理完了. 关掉所有的连接
    # 之所以在这进行这个操作是因为主 channel 也需要操作
    for my $c (@{ $cmgr->{channels} }) {
        next unless defined $c;
        if ($c->{wfd} &&
            $c->{extended}->length == 0 &&
            $c->{output}->length == 0 &&
            $c->{ostate} == CHAN_OUTPUT_WAIT_DRAIN ) { 
                $c->obuf_empty;
        }
        # 上面 obuf_empty 会给 ostate 变成 CHAN_OUTPUT_CLOSED
        # 下面这个就会发关闭给远程
        if ($c->delete_if_full_closed) {
            defined $c->{cb} ? $c->{cb}->() : '';
            $cmgr->remove($c->{id});
        }
    }
        
    my $oc = grep { defined } @{ $cmgr->{channels} };
    return $ssh->client_loop unless $oc > 1;

    my $cv = AE::cv sub {
        my $result = shift->recv;
        delete $ssh->{watcher};
        $ssh->event_loop($cmgr, $h, $cb);
    };

    # 这是处理频道上的输出, 客户端的输入
    for my $c (@{ $cmgr->{channels} }) {
        next unless defined $c;
        my $id = $c->{id};
        if ($c->{rfd} && $c->{istate} == CHAN_INPUT_OPEN &&
            $c->{remote_window} > 0 &&
            $c->{input}->length < $c->{remote_window}) {
            $ssh->{watcher}{$id}{rfd} = AE::io $c->{rfd}, 0, sub {
                # 顺序记录 - 频道 - rfd
                my $buf;
                sysread $c->{rfd}, $buf, 8192;
                ($buf) = $buf =~ /(.*)/s;
                $c->send_data($buf);
                $cv->send('rfd');
                delete $ssh->{watcher}{$id}{rfd}
            };
        } 

        # 给内容输出
        if (defined $c->{wfd} &&
            $c->{ostate} == CHAN_OUTPUT_OPEN ||
            $c->{ostate} == CHAN_OUTPUT_WAIT_DRAIN) {
            if ($c->{output} and $c->{output}->length > 0) {
                $ssh->{watcher}{$id}{wfd} = AE::io $c->{wfd}, 1, sub {
                   if (my $r = $c->{handlers}{"_output_buffer"}) {
                       $r->{code}->( $c, $c->{output}, @{ $r->{extra} } );
                   }
                   $c->{local_consumed} += $c->{output}->length;
                   $c->{output}->empty;
                   $cv->send('wfd');
                    delete $ssh->{watcher}{$id}{wfd}
                }