Anonymous

my $on_connect = sub {
		my ($port, $fh, $host) = @_;

		return close $fh unless exists $self->allowed_ip_hash->{$host};

		$self->{connections}->{$port}   = [
			grep { $_ && $_->[0x00] eq 'AnyEvent::Handle' } @{ $self->{connections}->{$port} // [] }
		];

		my $connect_no = @{ $self->{connections}->{$port} };

		return close $fh if $connect_no >= $self->threads;

		my $on_destroy_cb = sub {
			foreach ( @{$self->{connections}->{$port}->[$connect_no]} ) {
				close $_->fh if $_->fh;
				$_->destroy;
			}
			delete $self->{connections}->{$port}->[$connect_no];
			delete $self->proxies->{$port} unless @{$self->{connections}->{$port}};
		};

		$self->{connections}->{$port}->[$connect_no]->[0x00] = new AnyEvent::Handle
			fh			=> $fh,
			timeout		=> $self->timeout,
			on_eof		=> $on_destroy_cb,
			on_error	=> $on_destroy_cb,
			on_read		=> sub {
				$_[0x00]->{_my_rbuf} .= delete $_[0x00]{rbuf};
			}
		;

		my $max_proxy_av = $self->threads - $connect_no;

		$self->proxy_pull->GetProxy($self->proxies->{$port}, sub {

			return $on_destroy_cb->() unless $_[0x00];

			$self->proxies->{$port} = $_[0x01];

			$self->{connections}->{$port}->[$connect_no]->[0x01] = new AnyEvent::Handle
				fh			=> $_[0x00],
				timeout		=> $self->timeout,
				on_eof		=> $on_destroy_cb,
				on_error	=> $on_destroy_cb,
			;
			my ($client_h, $proxy_h) = @{ $self->{connections}->{$port}->[$connect_no] };
			weaken $client_h;
			weaken $proxy_h;

			$client_h->on_read(sub {
				$proxy_h->push_write(delete $_[0x00]{rbuf});
			});
			$proxy_h->on_read(sub {
				$client_h->push_write(delete $_[0x00]{rbuf});
			});
			$proxy_h->push_write($client_h->{_my_rbuf}) if $client_h->{_my_rbuf};
		});
	};