view release on metacpan or  search on metacpan

lib/AnyMongo/Connection.pm  view on Meta::CPAN

    is       => 'rw',
    isa      => 'Str',
    required => 0,
);

has password => (
    is       => 'rw',
    isa      => 'Str',
    required => 0,
);

has connected => (
    isa => 'Bool',
    is  => 'rw',
    default  => 0,
);

has cv => (
    isa => 'AnyEvent::CondVar',
    is  => 'ro',
    lazy_build => 1,
    clearer  => 'clear_cv',
);

sub _build_cv {
    my ($self) = @_;
    AE::cv;
}

has _connection_error => (
    isa => 'Bool',
    is  => 'rw',
    default  => 0,
);


sub CLONE_SKIP { 1 }

sub BUILD { shift->_init }

sub _init {
    my ($self) = @_;
    eval "use ${_}" # no Any::Moose::load_class becase the namespaces already have symbols from the xs bootstrap
        for qw/AnyMongo::Database AnyMongo::Cursor AnyMongo::BSON::OID/;
    $self->_parse_servers();
    if ($self->auto_connect) {
        $self->connect;
        # if (defined $self->username && defined $self->password) {
        #     $self->authenticate($self->db_name, $self->username, $self->password);
        # }
    }

}

sub connect {
    my ($self,%args) = @_;
    return if $self->connected || $self->{_trying_connect};
    # warn "connect...\n";
    $self->{_trying_connect} = 1;
    #setup connection timeout watcher
    my $timer; $timer = AnyEvent->timer( after => 5 ,cb => sub {
        undef $timer;
        unless ( $self->connected ) {
            $self->{_trying_connect} = 0;
            $self->cv->croak('Failed to connect to any mongodb servers,timeout');
        }
    });
    $self->cv->begin( sub {
        undef $timer;
        if ($self->connected) {
            $self->_connection_error(0);
            shift->send;
        }
        else {
            shift->croak("Failed to connect to any mongodb servers");
        }
    });

    my $servers = $self->{mongo_servers};
    my $seed_queue = $self->{_seed_queue} = [ keys %{ $servers } ];
    my $seed_tried = $self->{_seed_tried} = {};
    while (!$self->connected && @{$seed_queue} ) {
        while (my $h = shift @{$seed_queue}) {
            $seed_tried->{$h} = 1;
            $self->_check_master($h);
        }
    }
    $self->cv->recv;
    $self->{_trying_connect} = 0;
    # warn "connect done.\n";
}


sub _set_master {
    my ($self,$server_id,$h) = @_;
    $self->master_handle($h);
    $self->{mongo_servers}->{$server_id}->{is_master} = 1;
    $self->{master_id} = $server_id;
    $self->connected(1);
    # warn "master_id:$server_id\n";
}


sub _is_master {
    my ($config) = @_;
    $config && (ref($config) ne 'SCALAR') && ($config->{ismaster});
}

sub _check_master {
    my ($self,$server_id,$only_self) = @_;
    my $guards = $self->{_guards};
    return if $guards->{$server_id} or $self->connected;
    my $connect_cv = AE::cv;
    $self->_connect_to_host($server_id, sub { $connect_cv->send(shift) });
    # FIXME: as run_command also call "recv", this is
    # the easiest way to fight with recursive blocking wait.
    # Though,it's not a problem when conjunction with Coro.
    my $handle = $connect_cv->recv;
    if ($handle) {
        my $config = $self->admin->run_command({ismaster => 1},$handle);
        # warn "check ismaster:$server_id\n";
        if (ref $config ne 'SCALAR') {
            # if this is a replica set & we haven't renewed the host list in 1 sec
            my $servers = $self->{mongo_servers};
            if ($config->{hosts} && time() > $self->ts) {
                foreach my $h (@{$config->{hosts}}) {
                    $self->_add_host($h);
                }
                $self->ts(time);

lib/AnyMongo/Connection.pm  view on Meta::CPAN

        }
    };
}

sub _parse_servers {
    my ($self) = @_;
    my $str = $self->host;
    $str = substr $self->host, 10 if $str =~ /^mongodb:\/\//;
    my @pairs = split ",", $str;
    my $servers = {};
    my $server_seeds_cnt = 0;
    for my $h (@pairs) {
        my ($host,$port) = split ':',$h;
        $port ||= 27017;
        $servers->{$host.':'.$port} = {
            connected => 0,
            handle => undef,
            host => $host,
            port => $port,
            is_master => 0,
        };
    }
    # $self->_servers($servers);
    $self->{mongo_servers} = $servers;
}

sub send_message {
    my ($self,$data,$hd) = @_;
    croak 'connection lost' unless $hd or $self->_check_connection;
    $hd ||= $self->master_handle;
    $hd->push_write($data);
}

sub _check_connection {
    my ($self) = @_;
    $self->connected or ($self->auto_reconnect and $self->connect);
    $self->connected;
}

sub recv_message {
    my ($self,$hd) = @_;
    my ($message_length,$request_id,$response_to,$op) = $self->_receive_header($hd);
    my ($response_flags,$cursor_id,$starting_from,$number_returned) = $self->_receive_response_header($hd);
    $self->_check_respone_flags($response_flags);
    my $results =  $self->_read_documents($message_length-36,$cursor_id,$hd);
    return ($number_returned,$cursor_id,$results);
}

sub _check_respone_flags {
    my ($self,$flags) = @_;
    if (($flags & REPLY_CURSOR_NOT_FOUND) != 0) {
        croak("cursor not found");
    }
}

sub receive_data {
    my ($self,$size,$hd) = @_;
    $hd ||= $self->master_handle;
    croak 'connection lost' unless $hd or $self->_check_connection;
    my $cv = AE::cv;
    my $timer; $timer = AnyEvent->timer( after => $self->query_timeout ,cb => sub {
        undef $timer;
        $cv->croak('receive_data timeout');
    });
    $hd->push_read(chunk => $size, sub {
        my ($hdl, $bytes) = @_;
        $cv->send($_[1]);
    });
    $hd->{cv} = $cv;
    my $data = $cv->recv;
    delete $hd->{cv};
    $data;
}


sub _receive_header {
    my ($self,$hd) = @_;
    my $header_buf = $self->receive_data(STANDARD_HEADER_SIZE,$hd);
    croak 'Short read for DB response header; length:'.length($header_buf) unless length $header_buf == STANDARD_HEADER_SIZE;
    return unpack('V4',$header_buf);
}

sub _receive_response_header {
    my ($self,$hd) = @_;
    my $header_buf = $self->receive_data(RESPONSE_HEADER_SIZE,$hd);
    croak 'Short read for DB response header' unless length $header_buf == RESPONSE_HEADER_SIZE;
    my ($response_flags) = unpack 'V',substr($header_buf,0,BSON_INT32);
    my ($cursor_id) = unpack 'j',substr($header_buf,BSON_INT32,BSON_INT64);
    my ($starting_from,$number_returned) = unpack 'V2',substr($header_buf,BSON_INT32+BSON_INT64);
    return ($response_flags,$cursor_id,$starting_from,$number_returned);
}

sub _read_documents {
    my ($self,$doc_message_length,$cursor_id,$hd) = @_;
    my $remaining = $doc_message_length;
    my $bson_buf;
    # do {
    #     my $buf_len = $remaining > 4096? 4096:$remaining;
    #     $bson_buf .= $self->receive_data($buf_len);
    #     $remaining -= $buf_len;
    # } while ($remaining >0 );
    $bson_buf = $self->receive_data($doc_message_length,$hd);
    return unless $bson_buf;
    # warn "#_read_documents:bson_buf size:".length($bson_buf);
    # my $docs = decode_bson_documents($bson_buf,length($bson_buf));
    # warn '#_read_documents decode_bson_documents ...';
    my $docs = decode_bson_documents($bson_buf);
    # warn "docs:$docs";
    # warn "#_read_documents:".Dumper($docs)."\n";
    return $docs;
}

sub database_names {
    my ($self) = @_;
    my $ret = $self->admin->run_command({ listDatabases => 1 });
    return map { $_->{name} } @{ $ret->{databases} };
}


sub get_database {
    my ($self, $database_name) = @_;
    return AnyMongo::Database->new(