from .._protocol.incoming cimport DNSIncoming

from .._updates cimport RecordUpdateListener

from .._utils.time cimport current_time_millis

from .._record_update cimport RecordUpdate

cdef cython.float _DNS_PTR_MIN_TTL

cdef cython.uint _TYPE_PTR

cdef object _ADDRESS_RECORD_TYPES

cdef object RecordUpdate

cdef bint TYPE_CHECKING

cdef object _TYPE_PTR

record=DNSRecord,

answers=cython.list,

maybe_entry=DNSRecord,

rec_update=RecordUpdate

)

cpdef void async_updates_from_response(self, DNSIncoming msg)

address_adds.append(record)

else:

other_adds.append(record)

updates.append(RecordUpdate(record, maybe_entry))

rec_update = RecordUpdate.__new__(RecordUpdate)

rec_update._fast_init(record, maybe_entry)

updates.append(rec_update)

# This is likely a goodbye since the record is

# expired and exists in the cache

elif maybe_entry is not None:

updates.append(RecordUpdate(record, maybe_entry))

rec_update = RecordUpdate.__new__(RecordUpdate)

rec_update._fast_init(record, maybe_entry)

updates.append(rec_update)

removes.add(record)

if unique_types:

cdef public DNSRecord new

cdef public DNSRecord old

cdef void _fast_init(self, object new, object old)

from ._dns import DNSRecord

_DNSRecord = DNSRecord

class RecordUpdate:

__slots__ = ("new", "old")

def __init__(self, new: DNSRecord, old: Optional[DNSRecord] = None):

def __init__(self, new: DNSRecord, old: Optional[DNSRecord] = None) -> None:

"""RecordUpdate represents a change in a DNS record."""

self._fast_init(new, old)

def _fast_init(self, new: _DNSRecord, old: Optional[_DNSRecord]) -> None:

"""Fast init for RecordUpdate."""

self.new = new

self.old = old