ó
}�9Rc@s£dZd
dl
Z
d
dlZd
dlmZ
d
dlmZmZ
d
dlm	Z	m
Z

d
dlmZ
d
dl
mZmZ
def
d	„ƒYZd
„ZdS(sâManage outgoing and incoming messages when communicating with the server.

The protocol to communicate between the client and the server has been designed
to be very robust so that messages are not lost. In addition it is (vaguely)
symmetric, as the client and server need to send messages both ways.

Client->Server Payload
======================

All message payloads are bpickled with L{landscape.lib.bpickle.dumps}. Client
to server payloads are C{dict}s of the form::

  {'server-api': SERVER_API_VERSION,
   'client-api': CLIENT_API_VERSION,
   'sequence': SEQUENCE_NUMBER,
   'accepted-types': SERVER_ACCEPTED_TYPES_DIGEST,
   'messages': MESSAGES,
   'total-messages': TOTAL_COUNT_OF_PENDING_MESSAGES,
   'next-expected-sequence': EXPECTED_SEQUENCE_NUMBER,
   'client-accepted-types': CLIENT_ACCEPTED_TYPES (optional)}

The values have the following semantics:

  - C{SERVER_API_VERSION}: The API version that is required on the server
    in order to process the messages in this payload (the schema and semantics
    of message types are usually different for different API versions).

  - C{CLIENT_API_VERSION}: The API version of the client, hinting the server
    about the schema and semantics of the messages types accepted by the client
    (see below).

  - C{SEQUENCE_NUMBER}: A monotonically increasing nonnegative integer. The
    meaning of this is described below.

  - C{SERVER_ACCEPTED_TYPES_DIGEST}: A hash of the message types that the
    client thinks are currently accepted by the server. The server can use it
    to know whether to send the client a new up-to-date list of accepted
    message types.

  - C{MESSAGES}: A python list of messages, described below.

  - C{TOTAL_COUNT_OF_PENDING_MESSAGES}: The total number of messages in the
    client outgoing queue. This is includes the number of messages being sent
    in this payload, plus any other messages still pending and not included
    here.

  - C{EXPECTED_SEQUENCE_NUMBER}: The sequence number which the client expects
    the next message sent from the server to have.

  - C{CLIENT_ACCEPTED_TYPES}: Optionally, a list of message types that the
    client accepts. The server is supposed to send the client only messages of
    this type. It will be inlcuded in the payload only if the hash that the
    server sends us is out-of-date. This behavior is simmetric with respect to
    the C{SERVER_ACCEPTED_TYPES_DIGEST} field described above.

Server->Client Payload
======================

The payloads that the server sends to not-yet-registered clients (i.e. clients
that don't provide a secure ID associated with a computer) are C{dict}s of the
form::

  {'server-uuid': SERVER_UUID,
   'messages': MESSAGES}

where:

  - C{SERVER_UUID}: A string identifying the particular Landscape server the
    client is talking to.

  - C{MESSAGES}: A python list of messages, described below.

Additionally, payloads to registered clients will include these fields::

  {'next-expected-sequence': EXPECTED_SEQUENCE_NUMBER,
   'client-accepted-types-hash': CLIENT_ACCEPTED_TYPES_DIGEST,

where:

  - C{EXPECTED_SEQUENCE_NUMBER}: The sequence number which the server expects
    the next message sent from the client to have.

  - C{CLIENT_ACCEPTED_TYPES_DIGEST}: A hash of the message types that the
    server thinks are currently accepted by the client. The client can use it
    to know whether to send to the server an up-to-date list the message types
    it now accepts (see CLIENT_ACCEPTED_TYPES in the client->server payload).

Individual Messages
===================

A message is a C{dict} with required and optional keys. Messages are packed
into Python lists and set as the value of the 'messages' key in the payload.

The C{dict} of a single message is of the form::

  {'type': MESSAGE_TYPE,
   ...}

where:

  - C{MESSAGE_TYPE}: A simple string, which lets the server decide what handler
    to dispatch the message to, also considering the SERVER_API_VERSION value.

  - C{...}: Other entries specific to the type of message.

Message Sequencing
==================

A message numbering system is built in to the protocol to ensure robustness of
client/server communication. The way this works is not totally symmetrical, as
the client must connect to the server via HTTP, but the ordering that things
happen in over the course of many connections remains the same (see also
L{landscape.broker.store} for more concrete examples):

  - Receiver tells Sender which sequence number it expects the next batch of
    messages to start with.

  - Sender gives some messages to Receiver, specifying the sequence number of
    the first message. If the expected and actual sequence numbers are out of
    synch, Sender resynchronizes in a certain way.

The client and server must play the part of *both* of these roles on every
interaction, but it simplifies things to talk about them in terms of a single
role at a time.

When the client connects to the server, it does the following things acting
in the role of Sender (which is by far its more burdened role):

  - Send a payload containing messages and a sequence number. The sequence
    number should be the same number that the server gave as
    next-expected-sequence in the prior connection, or 0 if there was no
    previous connection.

  - Get back a next-expected-sequence from the server. If that value is is not
    len(messages) + previous-next-expected, then resynchronize.

It does the following when acting as Receiver:

  - Send a payload containing a next-expected-sequence, which should be the
    sequence number of the first message that the server responds with. This
    value should be previous-next-expected + len(previous_messages).

  - Receive some messages from the server, and process them immediately.

When the server is acting as Sender, it does the following:

  - Wait for a payload with next-expected-sequence from the client.

  - Perhaps resynchronize if next-expected-sequence is unexpected.

  - Respond with a payload of messages to the client. No sequence identifier
    is given for this payload of messages, because it would be redundant with
    data that has already passed over the wire (received from the client)
    during the very same TCP connection.

When the server is acting as a Receiver, it does the following:

  - Wait for a payload with a sequence identifier and a load of messages.
  - Respond with a next-expected-sequence.

There are two interesting exceptional cases which must be handled with
resynchronization:

  1. Messages received with sequence numbers less than the next expected
     sequence number should be discarded, and further messages starting at
     the expected sequence numbers should be processed.

  2. If the sequence number is higher than what the receiver expected, then
     no messages are processed and the receiver responds with the same
     {'next-expected-sequence': N}, so that the sender can resynchronize
     itself.

This implies that the receiver must record the sequence number of the last
successfully processed message, in order for it to respond to the sender
with that number. In addition, the sender must save outbound messages even
after they have been delivered over the transport, until the sender receives
a next-expected-sequence higher than the outbound message. The details of
this logic are described in L{landscape.broker.store}.
iÿÿÿÿN(
tmd5(tDeferredtsucceed(tgot_next_expectedtANCIENT(
tformat_delta(t
SERVER_APIt
CLIENT_APItMessageExchangecBs×eZ
dZd
d„
Zd„Zed„
Zd„Zd„Zd„Z	d„Z
d	„Zd
„Zd„Z
d„Zeed
„Zd„Zd„Zd„Zd„Zd„Zd„Zd„Zd„Zd„ZRS(svSchedule and handle message exchanges with the server.

    The L{MessageExchange} is the place where messages are sent to go out
    to the Landscape server. It accumulates messages in its L{MessageStore}
    and periodically delivers them to the server.

    It is also the place where messages coming from the server are handled. For
    each message type the L{MessageExchange} supports setting an handler that
    will be invoked when a message of the that type is received.

    An exchange is performed with an HTTP POST request, whose body contains
    outgoing messages and whose response contains incoming messages.
    idcCsò|
|_||_
||_||_||_|j|_|j|_||_	d|_d|_t
|_t
|_tƒ|_d|_i|_||_t
|_|jd
|jƒ
|jd|jƒ
|jd|jƒ
|
jd|jƒ
dS(s³
        @param reactor: The L{LandscapeReactor} used to fire events in response
            to messages received by the server.
        @param store: The L{MessageStore} used to queue outgoing messages.
        @param transport: The L{HTTPTransport} used to deliver messages.
        @param registration_info: The L{Identity} storing our secure ID.
        @param config: The L{BrokerConfiguration} with the `exchange_interval`
            and `urgent_exchange_interval` parameters, respectively holding
            the time interval between subsequent exchanges of non-urgent
            messages, and the time interval between subsequent exchanges
            of urgent messages.
        saccepted-typest
resynchronizes
set-intervalssresynchronize-clientsN(t_reactort_message_storet
_transportt_registration_infot_configtexchange_intervalt_exchange_intervalturgent_exchange_intervalt_urgent_exchange_intervalt
_max_messagestNonet_notification_idt_exchange_idtFalset_exchangingt_urgent_exchangetsett_client_accepted_typest_client_accepted_types_hasht_message_handlerst_exchange_storet_stoppedtregister_messaget_handle_accepted_typest_handle_resynchronizet_handle_set_intervalstcall_ont_resynchronize(tselftreactortstoret	transporttregistration_infotexchange_storetconfigtmax_messages((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyt__init__Îs*	
	
	
	
	


	
	
	
	
	

	
	
	
	


cCspd
|
krtS|
d
}|j
j|ƒ
}|dkrMtjd|ƒ

tS|jj|jk}|jƒ
|S(s’Returns C{True} if message is obsolete.

        A message is considered obsolete if the secure ID changed since it was
        received.
        soperation-ids4No message context for message with operation-id: %sN(	RRtget_message_contextRtloggingtwarningR
t	secure_idtremove(R&tmessagetoperation_idtcontexttresult((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyt_message_is_obsoleteós









cCs„|j|
ƒ
r-t
jd
|
jdƒ
ƒ

dSd|
krUt|jjƒƒ
|
d<n|jj	|
ƒ
}|r€|j
dtƒ

n|S(sÊInclude a message to be sent in an exchange.

        If urgent is True, an exchange with the server will be
        scheduled urgently.

        @param message: Same as in L{MessageStore.add}.
        snResponse message with operation-id %s was discarded because the client's secure ID has changed in the meantimesoperation-idt	timestampturgentN(R8R0tinfotgetRtintR
ttimeRtaddtschedule_exchangetTrue(R&R4R:t
message_id((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pytsend
s







c

Cs|jd
t
ƒ

dS(s9Start scheduling exchanges. The first one will be urgent.R:N(R@RA(
R&((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pytstart!
sc

Csi|jd
k	r.|jj|jƒ

d
|_n|jd
k	r\|jj|jƒ

d
|_nt|_d
S(sStop scheduling exchanges.N(RRR
tcancel_callRRAR(
R&((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pytstop%
s



cCsÓt|j
jƒƒ
}t|
d
ƒ
}t||ƒ}|j
j|ƒ

tjd|ƒ
|j
jdƒ
ry|jdt	ƒ

nx(||D]}|j
jd|tƒ
q„Wx(||D]}|j
jd|t	ƒ
q¯WdS(s‚

        When the server updates us about the types of message it
        accepts, update our message store.

        If this makes existing held messages available for sending,
        urgently exchange messages.

        If new types are made available or old types are dropped a
        C{("message-type-acceptance-changed", type, bool)} reactor
        event will be fired.
        ttypessAccepted types changed: %si
R:smessage-type-acceptance-changedN(
RRtget_accepted_typestget_accepted_types_difftset_accepted_typesR0R;tget_pending_messagesR@RAR
tfireR(R&R4t	old_typest	new_typestdiffttype((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR!1
s









cCs9|
d
}|jidd6|d
6ƒ

|j
jdƒ

dS(Nsoperation-idR	RPsresynchronize-clients(RCR
RL(R&R4topid((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR"I
s



c

Cs|jd
t
ƒ

dS(NR:(R@RA(
R&((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR%N
s
cCs}d
|
kr6|
d
|j_
tjd|jj
ƒ

nd|
krl|
d|j_tjd|jjƒ

n|jjƒ
dS(Ntexchanges$Exchange interval set to %d seconds.surgent-exchanges+Urgent exchange interval set to %d seconds.(RRR0R;Rtwrite(R&R4((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR#Q
s


	



	

c
	sæˆjrt
dƒ
Stˆ_ˆjjd
ƒ

ˆjƒ‰
tjƒ‰ˆjrjt	j
dˆjjƒƒ

nt	j
dˆjjƒƒ

t
ƒ‰‡‡‡‡
fd†}
ˆjj|
dˆjjˆ
ˆjjˆjƒˆ
jdƒ
ƒ
ˆS(s|
Send pending messages to the server and process responses.

        A C{pre-exchange} reactor event will be emitted just before the
        actual exchange takes place.

        An C{exchange-done} or C{exchange-failed} reactor event will be
        emitted after a successful or failed exchange.

        @return: A L{Deferred} that is fired when exchange has completed.
        spre-exchanges)Starting urgent message exchange with %s.s"Starting message exchange with %s.c

s²tˆ
_
|rDˆ
jr1tjd
ƒ

tˆ
_nˆ
jˆ|ƒ
nˆ
jjdƒ

tjdƒ

ˆ
jdt	ƒ

ˆ
jjdƒ

tjdt
tjƒˆƒ
ƒ
ˆjdƒ

dS(Ns"Switching to normal exchange mode.sexchange-failedsMessage exchange failed.tforces
exchange-dones!Message exchange completed in %s.(RRRR0R;t_handle_resultR
RLR@RARR>tcallbackR(
R7(t
start_timeR&tdeferredtpayload(s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyt
handle_resultz
s
	

	







	

s
server-apiN(RRRRAR
RLt
_make_payloadR>RR0R;Rtget_urlR
tcall_in_threadRRR
R2t_get_exchange_tokenR<(R&RZ((RWRYRXR&s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyRR\
s&	

	
	
	
	
	

	
	

c


Cs|jS(
sMReturn a bool showing whether there is an urgent exchange scheduled.
        (
R(
R&((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyt	is_urgent’
scCsý|jr
d
S|j
rù|s<|jd
ks<|
rù|jrù|
rNt|_n|jrm|jj|jƒ

n|jr…|jj	}n|jj
}|jd
k	r¶|jj|jƒ

n|d}|jj||j
ƒ|_|jj||jƒ|_nd
S(s_Schedule an exchange to happen.

        The exchange will occur after some time based on whether C{urgent} is
        True. An C{impending-exchange} reactor event will be emitted
        approximately 10 seconds before the exchange is started.

        @param urgent: If true, ensure an exchange happens within the
            urgent interval.  This will reschedule the exchange if necessary.
            If another urgent exchange is already scheduled, nothing happens.
        @param force: If true, an exchange will necessarily be scheduled,
            even if it was already scheduled before.
        Ni
(RRRRRRAR
RERRRRt
call_latert_notify_impending_exchangeRR(R&R:RTtintervaltnotification_interval((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR@—
s&
	






	
	




	

c
Cs0|jj
ƒ}
|jjd
ƒ

|jjƒ
|
S(s
Get the token given us by the server at the last exchange.

        It will be C{None} if we are not fully registered yet or if something
        bad happened during the last exchange and we could not get the token
        that the server had given us.
        N(Rtget_exchange_tokentset_exchange_tokenRtcommit(R&texchange_token((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR^¿
s

c

Cs|jj
d
ƒ

dS(Nsimpending-exchange(R
RL(
R&((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyRaÓ
s
c
CsC
|j}
|j
|
jƒƒ
}|
j|jƒ
}|
jƒ}|r¿|d
jdƒ
}x9t|ƒ
D]%\}}|jdƒ
|krbPqbqbWd}|dk	r§||3n|dkrÅd}qÅnt	}i|d6t
d6|
jƒd6|d6|d6|d	6|
jƒd
6}|j
ƒ}	|j
|	ƒ
}
|
|jkr?
|	|d<n|S(
sèReturn a dict representing the complete exchange payload.

        The payload will contain all pending messages eligible for
        delivery, up to a maximum of C{max_messages} as passed to
        the L{__init__} method.
        itapis2.0s
server-apis
client-apitsequencesaccepted-typestmessagesstotal-messagessnext-expected-sequencesclient-accepted-typesN(Rt_hash_typesRHRKRtcount_pending_messagesR<t	enumerateRRRtget_sequencetget_server_sequencet!get_client_accepted_message_typesR(R&R(taccepted_types_digestRjttotal_messagest
server_apit
iR4RYtaccepted_client_typestaccepted_client_types_hash((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR[Ö
s6	
























cCsd
j|
ƒ
}t
|ƒ
jƒS(Nt
;(tjoinRtdigest(R&RGtaccepted_types_str((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyRks

cCsÑ
|j}|j
d
ƒ
|_|j
dƒ
}|jƒ}|dkre|jƒ}|t|
dƒ
7}nt||ƒ}|tkr´tj	dƒ

|j
i
dd6ƒ

|jjdƒ

n|j
|j
dƒ
ƒ

|jƒ}|j
d	ƒ
}||kr.
tj	d
||fƒ

|jjd||ƒ
|j|ƒ

n|jƒ
|jƒ}	xH|j
ddƒD]4}
|j|
ƒ

|	d7}	|j|	ƒ

|jƒ
qW
W|jdƒ
rÍ
tj	d
ƒ

||krÍ
|jdtƒ

qÍ
ndS(sÀ
Handle a response from the server.

        Called by L{exchange} after a batch of messages has been
        successfully delivered to the server.

        If the C{server_uuid} changed, a C{"server-uuid-changed"} event
        will be fired.

        Call L{handle_message} for each message in C{result}.

        @param payload: The payload that was sent to the server.
        @param result: The response got in reply to the C{payload}.
        sclient-accepted-types-hashsnext-expected-sequenceRjsIServer asked for ancient data: resynchronizing all state with the server.R	RPsresynchronize-clientssnext-exchange-tokensserver-uuids%Server UUID changed (old=%s, new=%s).sserver-uuid-changedi
s0Pending messages remain after the last exchange.R:N((RR<RRnRtlenRRR0R;RCR
RLRetget_server_uuidtset_server_uuidRfRothandle_messagetset_server_sequenceRKR@RA(R&RYR7t
message_storet
next_expectedtold_sequencetmessage_store_statetold_uuidtnew_uuidRiR4((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyRUs@	











	














cCs0|jj
|
gƒj|ƒ

|jj|
ƒ

d
S(s
Register a handler for the given message type.

        The C{handler} callable will to be executed when a message of
        type C{type} has been received from the server.

        Multiple handlers for the same type will be called in the
        order they were registered.
        N(Rt
setdefaulttappendRR?(R&RPthandler((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR Ls	
cCs†d
|
kr3|jj
|
d
|jj|
dƒ
n|jjd|
ƒ
|
d|jkr‚x&|j|
dD]}||
ƒ

qkWndS(s–
        Handle a message received from the server.

        Any message handlers registered with L{register_message} will
        be called.
        soperation-idRPR4N(Rtadd_message_contextR
R2R
RLR(R&R4Rˆ((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR~Xs	



cCs|jj
t|
ƒ
ƒ

dS(
N(RR?tstr(R&RP((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyt%register_client_accepted_message_typelsc

Cs
t|j
ƒ
S(
N(tsortedR(
R&((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyRpqs
(t__name__t
__module__t__doc__R.R8RRCRDRFR!R"R%R#RRR_R@R^RaR[RkRUR R~R‹Rp(((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyR¿s,
$								6	(			.		D			cCsµt|ƒ
}t|
ƒ
}
|
|}||
@}||
}g}|j
g|D]}d
|^qIƒ

|j
g|D]}d|^qmƒ

|j
g|D]}d|^q‘ƒ

dj|ƒ
S(Ns+%ss%ss-%st
 (RtextendRx(RMRNtadded_typeststable_typest
removed_typesRORP((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyRIus









$
$
$
(R�R>R0tlandscape.lib.hashlibRttwisted.internet.deferR
Rtlandscape.lib.messageRRt
landscape.logRt	landscapeRRtobjectRRI(((s=/usr/lib/python2.7/dist-packages/landscape/broker/exchange.pyt<module>³s




ÿ·