�
}�9Rc@s�dZd
dl
Z
d
dlZd
dlZd
dlZd
dlmZ
d
dlmZ
dZ	dZ
def
d��YZd	�Z
dS(
s�Message storage.

The sequencing system we use in the message store may be quite confusing
if you haven't looked at it in the last 10 minutes. For that reason, let's
review the mechanics here.

Our goal is to implement a reasonably robust system for delivering messages
from us to our peer. The system should be smart enough to recover if the peer
happens to lose messages that we have already sent, provided that these
messages are not too old (we'll see below what 'too old' means).

Messages added to the store are identified by increasing natural numbers, the
first message added is identified by 0, the second by 1, and so on. We call
"sequence" the number identifying the next message that we want to send. For
example, if the store has been added ten messages (that we represent with
uppercase letters) and we want start sending the first of them, our store
would like like::

    sequence: 0
    messages: A, B, C, D, E, F, G, H, I, J
              ^

The "^" marker is what we call "pending offset" and is the displacement of the
message we want to send next from the first message we have in the store.

Let's say we now send to our peer a batch of 3 sequential messages. In the
payload we include the body of the messages being sent and the sequence, which
identifies the first message of the batch. In this case the payload would look
like (pseudo-code)::

    (sequence: 0, messages: A, B, C)

If everything works fine on the other end, our peer replies with a payload that
would like::

    (next-expected-sequence: 4)

meaning that the peer as received all the three messages that we sent, an so
the next message it expects to receive is the one identified by the number 4.
At this point we update both our pending offset and our sequence values, and
the store now looks like::

    sequence: 4
    messages: A, B, C, D, E, F, G, H, I, J
                       ^

Great, now let's pretend that we send another batch, this time with five
messages::

    (sequence: 4, messages: D, E, F, G, H)

Our peer receives them fine responding with a payload looking like::

    (next-expected-sequence: 9)

meaning that it received all the eight messages we sent so far and it's waiting
for the ninth. This is the second successful batch that we send in a row, so we
can be reasonably confident that at least the messages in the first batch are
not really needed anymore. We delete them and we update our sequence and
pending offset accordingly::

    sequence: 9
    messages: D, E, F, G, H, I, J
                             ^

Note that we still want to keep around the messages we sent in the very last
batch, just in case. Indeed we now try to send a third batch with the last two
messages that we have, but our peer surprisingly replies us with this payload::

    (next-expected-sequence: 6)

Ouch! This means that something bad happened and our peer has somehow lost not
only the two messages that we sent in the last batch, but also the last three
messages of the former batch :(

Luckly we've kept enough old messages around that we can try to send them
again, we update our sequence and pending offset and the store looks like::

    sequence: 6
    messages: D, E, F, G, H, I, J
                    ^

We can now start again sending messages using the same strategy.

Note however that in the worst case scenario we could receive from our peer
a next-expected-sequence value which is so old to be outside our buffer
of already-sent messages. In that case there is now way we can recover the
lost messages, and we'll just send the oldest one that we have.

See L{MessageStore} for details about how messages are stored on the file
system and L{landscape.lib.message.got_next_expected} to check how the
strategy for updating the pending offset and the sequence is implemented.
i����N(
tbpickle(
t
SERVER_APIt
ht
btMessageStorecBsO
eZ
dZeZd
ejd�Zd�Zd�Zd�Z	d�Z
d�Zd�Zd	�Z
d
�Zd�Zd�Zd
�Zd�Zd�Zd�Zd�Zd�Zd$d�
Zd�Zd�Zd�Zd�Zd�Zd�Zd�Zd$d�
Z dd�
Z!d�Z"d�Z#d �Z$d!�Z%d"�Z&d#�Z'RS(%sRA message store which stores its messages in a file system hierarchy.

    Beside the "sequence" and the "pending offset" values described in the
    module docstring above, the L{MessageStore} also stores what we call
    "server sequence", which is the next message number expected by the
    *client* itself (because we are in turn the peer of a specular message
    system running in the server, which tries to deliver messages to us).

    The server sequence is entirely unrelated to the stored messages, but is
    incremented when successfully receiving messages from the server, in the
    very same way described above but with the roles inverted.

    @param persist: a L{Persist} used to save state parameters like the
        accepted message types, sequence, server uuid etc.
    @param directory: base of the file system hierarchy
    i�cCsq||_||_
||_i|_|
|_|
jd
�
|_|j�}tj	j
|�
smtj|�

ndS(Ns
message-store(t	_get_timet
_directoryt_directory_sizet_schemast_original_persisttroot_att_persistt_message_dirtostpathtisdirtmakedirs(tselftpersistt	directorytdirectory_sizetget_timetmessage_dir((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt__init__�s	
	
	
	
	



c


Cs|jj
�
d
S(sPersist metadata to disk.N(R	tsave(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytcommit�scCsNt|
�
t
ttfks!t�
|jjd
tt|
�
�
�
|j�
dS(s�Specify the types of messages that the server will expect from us.

        If messages are added to the store which are not currently
        accepted, they will be saved but ignored until their type is
        accepted.
        saccepted-typesN(ttypettupletlisttsettAssertionErrorRtsortedt_reprocess_holding(Rttypes((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_accepted_types�s!

c

Cs|jj
d
d�S(s)Get a list of all accepted message types.saccepted-types((Rtget(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_accepted_types�scCs|
|j�kS(
s>Return bool indicating if C{type} is an accepted message type.(
R$(RR((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytaccepts�sc

Cs|jj
d
d�S(s�Get the current sequence.

        @return: The sequence number of the message that the server expects us
            to send on the next exchange.
        tsequencei(RR#(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_sequence�scCs|jj
d
|
�
dS(s�Set the current sequence.

        Set the sequence number of the message that the server expects us to
        send on the next exchange.
        R&N(RR(Rtnumber((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_sequence�sc

Cs|jj
d
d�S(s�Get the current server sequence.

        @return: the sequence number of the message that we will ask the server
            to send to us on the next exchange.
        tserver_sequencei(RR#(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_server_sequence�scCs|jj
d
|
�
dS(s�Set the current server sequence.

        Set the sequence number of the message that we will ask the server to
        send to us on the next exchange.
        R*N(RR(RR(((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_server_sequence�sc

Cs|jj
d
�
S(s%Return the currently set server UUID.tserver_uuid(RR#(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_server_uuid�scCs|jj
d
|
�
dS(s=Change the known UUID from the server we're communicating to.R-N(RR(Rtuuid((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_server_uuid�sc

Cs|jj
d
�
S(s:Get the authentication token to use for the next exchange.texchange_token(RR#(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_exchange_token�scCs|jj
d
|
�
dS(s:Set the authentication token to use for the next exchange.R1N(RR(Rttoken((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_exchange_token�sc

Cs|jj
d
d�S(sGet the current pending offset.tpending_offseti(RR#(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_pending_offset�scCs|jj
d
|
�
dS(s�Set the current pending offset.

        Set the offset into the message pool to consider assigned to the
        current sequence number as returned by l{get_sequence}.
        R5N(RR(Rtval((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_pending_offset�scCs|j|j
�|
�

d
S(s/Increment the current pending offset by C{val}.N(R8R6(RR7((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytadd_pending_offset�sc

Cstd
�|j
�D�
�
S(s&Return the number of pending messages.c
ss|]}
dV
qd
S(i
N((t.0t
x((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pys	<genexpr>�s(tsumt_walk_pending_messages(
R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytcount_pending_messages�scCs�|j�}g}x�|j
�D]�}|
dk	rGt|�
|
krGPn|j|j|�
�
}ytj|�
}Wn0tk
r�
}
t	j
|�

|j|t�
qX|d
|kr�|j|t
�
q|j|�

qW|S(s;Get any pending messages that aren't being held, up to max.RN(R$R=tNonetlent_get_contentRRtloadst
ValueErrortloggingt	exceptiont
_add_flagstBROKENtHELDtappend(Rtmaxtaccepted_typestmessagestfilenametdatatmessaget
e((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_pending_messages�s












c
Csyxrtj
|jd
tt�
|j��D]H}
tj|
�

tjj	|
�
d}tj
|�
s)tj|�

q)q)WdS(s>Delete messages which are unlikely to be needed in the future.texcludeiN(t	itertoolstislicet_walk_messagesRHRGR6R
tunlinkRtsplittlistdirtrmdir(Rtfntcontaining_dir((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytdelete_old_messages�s





c
Cs5|jd
�

x!|j
�D]}
tj|
�

qWdS(sRemove ALL stored messages.iN(R8RUR
RV(RRM((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytdelete_all_messages
s


cCs|
|j|
j
<d
S(s�Add a schema to be applied to messages of the given type.

        The schema must be an instance of L{landscape.schema.Message}.
        N(RR(Rtschema((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt
add_schema

scCs�d
}|j�}x�|j
dt�
D]n}|j|�
}t|ksR||krntj|�
j|
krntSt|kr%t|kr%|d7}q%q%Wt	S(s�Return bool indicating if C{message_id} still hasn't been delivered.

        @param message_id: Identifier returned by the L{add()} method.
        iRRi
(
R6RURGt
_get_flagsRHR
tstattst_inotTruetFalse(Rt
message_idt
iR5RMtflags((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt
is_pending
s








cCs�d
|
kst�
|j
|
d
j|
�
}
d|
krH|j|
d<ntj|
�
}|j�}t|dd�}|j|�

|j	�
t
j|d|�
|j|
d
�
s�|j
|t�}nt
j|�
j}|S(s
Queue a message for delivery.

        @param message: a C{dict} with a C{type} key and other keys conforming
            to the L{Message} schema for that specifc message type.
        @return: message_id, which is an identifier for the added message.
        Rtapis.tmpt
w(RRtcoerceRiRtdumpst_get_next_message_filenametopentwritetcloseR
trenameR%t
_set_flagsRHRaRb(RROtmessage_dataRMtfileRe((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytadd!
s







c
Cs�|j�}
|
r|
d
}nt
j|jd�
�

d}|j|�
}|se|j|d�}n�t|�
|jkr�tt|d
jd�
d�
d�
}|j||�}nA|jtt|�
d�
�
}t
j|�

t
j	j
|d�}|S(Ni����t
0t
_ii
(t_get_sorted_filenamesR
RRR@RtstrtintRWRtjoin(Rtmessage_dirst
newest_dirtmessage_filenamesRM((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRmD
s








'



c
csQ|j�}
x>t
|jd
tt�
�
D] \}}||
kr)|V
q)q)WdS(s,Walk the files which are definitely pending.RRN(R6t	enumerateRURHRG(RR5RfRM((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR=Y
s




ccs�|
rt|
�
}
n|j
�}xc|D][}xR|j
|�
D]A}t|j|�
�
}|
sk|
|@r>|j||�V
q>q>Wq(WdS(
N(RRxR`R(RRRR|RRMRg((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRUa
s








tcCsQgtj
|j|
�
�
D]}|jd
�
s|^q}|jdd��

|S(Ns.tmptkeyc

Sst|j
d
�
d�
S(NRwi(RzRW(
R;((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt<lambda>n
s(R
RXRtendswithtsort(RtdirR;t
message_files((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRxk
s



c
Gstj
j|j|
�
S(
N(R
RR{R(Rtargs((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRq
s
cCs,t|
�
}z|j
�SWd|j�
XdS(
N(RntreadRp(RRMRt((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRAt
s


c

CsD
d
}
|j�}|j
�}x
|j�D]
}|j|�
}ytj|j|�
�
}Wn9tk
r�
}
tj	|�

t
|kr<
|
d7}
q<
q+X|d|k}t
|kr�|r<
|j�}	tj
||	�
|j|	t|�
tt
�
�
q<
q+|r2
|
|kr2
|j|t|�
tt
�
B�
n|
d7}
q+WdS(se
        Unhold accepted messages left behind, and hold unaccepted
        pending messages.
        ii
RN(R6R$RUR`RRBRARCRDRERHRmR
RqRrR(
RtoffsetR5RKtold_filenameRgRORPtacceptedtnew_filename((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR {
s(















&
#
cCs3tj
j|
�
}d
|kr/|jd
�
dSdS(NRwi
R�(R
RtbasenameRW(RRR�((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR`�
s



cCsztj
j|
�
\}}tj
j||jd
�
d�}|rf|d
djtt|�
�
�
7}ntj|
|�
|S(NRwiR�(R
RRWR{RRRq(RRRgtdirnameR�tnew_path((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRr�
s

"

&

cCs!|j|
|j
|
�
|�
dS(
N(RrR`(RRRg((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRF�
s
N((t__name__t
__module__t__doc__R
RittimeRRR"R$R%R'R)R+R,R.R0R2R4R6R8R9R>R?RQR\R]R_RhRuRmR=RURxRRAR R`RrRF(((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRlsH
																						#		
					cOsGd
dlm
}
t||
�}x!|j�D]}|j|�

q,W|S(sP
    Get a L{MessageStore} object with all Landscape message schemas added.
    i����(
tmessage_schemas(tlandscape.message_schemasR�RtvaluesR_(R�tkwargsR�tstoreR^((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_default_message_store�
s




(R�R�RSRDR
t
landscape.libRt	landscapeR
RHRGtobjectRR�(((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt<module>]s




�>