view release on metacpan or  search on metacpan

lib/API/MikroTik.pm  view on Meta::CPAN

    $_->{timeout} && $_->{loop}->remove($_->{timeout})
        for values %{$self->{requests}};
    $_ && $_->unsubscribe('close')->close() for values %{$self->{handles}};
    delete $self->{handles};
}

sub _close {
    my ($self, $loop) = @_;
    $self->_fail_all($loop, 'closed prematurely');
    delete $self->{handles}{$loop};
    delete $self->{responses}{$loop};
}

sub _command {
    my ($self, $loop, $cmd, $attr, $query, $cb) = @_;

    my $tag = ++$self->{_tag};
    my $r = $self->{requests}{$tag} = {tag => $tag, loop => $loop, cb => $cb};
    $r->{subscription} = delete $attr->{'.subscription'};

    warn "-- got request for command '$cmd' (tag: $tag)\n" if DEBUG;

    $r->{sentence} = encode_sentence($cmd, $attr, $query, $tag);
    return $self->_send_request($r);
}

sub _connect {
    my ($self, $r) = @_;

    warn "-- creating new connection\n" if DEBUG;

    my $queue = $self->{queues}{$r->{loop}} = [$r];

    my $tls = $self->tls;
    my $port = $self->port ? $self->{port} : $tls ? 8729 : 8728;

    $r->{loop}->client(
        {
            address     => $self->host,
            port        => $port,
            timeout     => CONN_TIMEOUT,
            tls         => $tls,
            tls_ciphers => 'HIGH'
        } => sub {
            my ($loop, $err, $stream) = @_;

            delete $self->{queues}{$loop};

            if ($err) { $self->_fail($_, $err) for @$queue; return }

            warn "-- connection established\n" if DEBUG;

            $self->{handles}{$loop} = $stream;

            weaken $self;
            $stream->on(read => sub { $self->_read($loop, $_[1]) });
            $stream->on(
                error => sub { $self and $self->_fail_all($loop, $_[1]) });
            $stream->on(close => sub { $self && $self->_close($loop) });

            $self->_login(
                $loop,
                sub {
                    if ($_[1]) {
                        $_[0]->_fail($_, $_[1]) for @$queue;
                        $stream->close();
                        return;
                    }
                    $self->_write_sentence($stream, $_) for @$queue;
                }
            );
        }
    );

    return $r->{tag};
}

sub _enqueue {
    my ($self, $r) = @_;
    return $self->_connect($r) unless my $queue = $self->{queues}{$r->{loop}};
    push @$queue, $r;
    return $r->{tag};
}

sub _fail_all {
    $_[0]->_fail($_, $_[2])
        for grep { $_->{loop} eq $_[1] } values %{$_[0]->{requests}};
}

sub _finish {
    my ($self, $r, $err) = @_;
    delete $self->{requests}{$r->{tag}};
    if (my $timer = $r->{timeout}) { $r->{loop}->remove($timer) }
    $r->{cb}->($self, ($self->{error} = $err // ''), $r->{data});
}

sub _login {
    my ($self, $loop, $cb) = @_;
    warn "-- trying to log in\n" if DEBUG;

    $loop->delay(
        sub {
            $self->_command($loop, '/login', {}, undef, $_[0]->begin());
        },
        sub {
            my ($delay, $err, $res) = @_;
            return $self->$cb($err) if $err;
            my $secret
                = md5_sum("\x00", $self->password, pack 'H*', $res->[0]{ret});
            $self->_command($loop, '/login',
                {name => $self->user, response => "00$secret"},
                undef, $delay->begin());
        },
        sub {
            $self->$cb($_[1]);
        },
    );
}

sub _read {
    my ($self, $loop, $bytes) = @_;

    warn "-- read bytes from socket: " . (length $bytes) . "\n" if DEBUG;

    my $response = $self->{responses}{$loop} ||= API::MikroTik::Response->new();
    my $data = $response->parse(\$bytes);

    for (@$data) {
        next unless my $r = $self->{requests}{delete $_->{'.tag'}};
        my $type = delete $_->{'.type'};
        push @{$r->{data} ||= Mojo::Collection->new()}, $_
            if %$_ && !$r->{subscription};

        if ($type eq '!re' && $r->{subscription}) {
            $r->{cb}->($self, '', $_);

        }
        elsif ($type eq '!done') {
            $r->{data} ||= Mojo::Collection->new();
            $self->_finish($r);

        }
        elsif ($type eq '!trap' || $type eq '!fatal') {
            $self->_fail($r, $_->{message});
        }
    }
}

sub _send_request {
    my ($self, $r) = @_;
    return $self->_enqueue($r) unless my $stream = $self->{handles}{$r->{loop}};
    return $self->_write_sentence($stream, $r);
}

sub _write_sentence {
    my ($self, $stream, $r) = @_;
    warn "-- writing sentence for tag: $r->{tag}\n" if DEBUG;

    $stream->write($r->{sentence});

    return $r->{tag} if $r->{subscription};

    weaken $self;
    $r->{timeout} = $r->{loop}
        ->timer($self->timeout => sub { $self->_fail($r, 'response timeout') });

    return $r->{tag};
}

1;