ó ½-'Nc@s…dZddlZddlZddlZddlZddlmZddlmZdZ dZ de fd„ƒYZ d „Z dS( sMessage storage.iÿÿÿÿN(tbpickle(t SERVER_APIthtbt MessageStorecBs@eZdZeZdd#ejd„Zd„Zd„Zd„Z d„Z d„Z d „Z d „Z d „Zd „Zd „Zd„Zd„Zd„Zd„Zd$d„Zd„Zd„Zd„Zd„Zd„Zd„Zd„Zd$d„Zdd„Zd„Z d„Z!d„Z"d „Z#d!„Z$d"„Z%RS(%sÍA message store which stores its messages in a file system hierarchy. 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 terminology here. Assume we have 10 messages in the store, which we label by the following uppercase letters:: A, B, C, D, E, F, G, H, I, J ^ Let's say that the next message we should send to the server is D. What we call "pending offset" is the displacement from the first message, which in our example above would be 3. What we call "sequence" is the number that the server expects us to label message D as. It could be pretty much any natural number, depending on the history of our exchanges with the server. What we call "server sequence", is the next message number expected by the *client* itself, and is entirely unrelated to the stored messages. ièi<cCsq||_||_||_i|_||_|jdƒ|_|jƒ}tj j |ƒsmtj |ƒndS(sÐ @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 s message-storeN( t _get_timet _directoryt_directory_sizet_schemast_original_persisttroot_att_persistt _message_dirtostpathtisdirtmakedirs(tselftpersistt directorytdirectory_sizetmonitor_intervaltget_timet message_dir((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt__init__)s      cCs|jjƒdS(sPersist metadata to disk.N(R tsave(R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytcommit:scCsNt|ƒtttfks!t‚|jjdtt|ƒƒƒ|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(ttypettupletlisttsettAssertionErrorR tsortedt_reprocess_holding(Rttypes((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_accepted_types>s!cCs|jjddƒS(s)Get a list of all accepted message types.saccepted-types((R tget(R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_accepted_typesIscCs||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.pytacceptsMscCs|jjddƒ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(R R$(R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt get_sequenceQscCs|jjd|ƒ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(R R(Rtnumber((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyt set_sequenceYscCs|jjddƒ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(R R$(R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_server_sequenceascCs|jjd|ƒ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(R R(RR)((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_server_sequenceiscCs|jjdƒS(s%Return the currently set server UUID.t server_uuid(R R$(R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_server_uuidqscCs|jjd|ƒdS(s=Change the known UUID from the server we're communicating to.R.N(R R(Rtuuid((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_server_uuiduscCs|jjddƒS(sGet the current pending offset.tpending_offseti(R R$(R((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_pending_offsetyscCs|jjd|ƒ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}. R2N(R R(Rtval((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytset_pending_offset}scCs|j|jƒ|ƒdS(s/Increment the current pending offset by C{val}.N(R5R3(RR4((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytadd_pending_offset…scCstd„|jƒDƒƒS(s&Return the number of pending messages.css|] }dVqdS(iN((t.0tx((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pys ‹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_typestmessagestfilenametdatatmessagete((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytget_pending_messagess  cCsyxrtj|jdttƒ|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_messagesRERDR3R tunlinkRtsplittlistdirtrmdir(Rtfntcontaining_dir((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytdelete_old_messages¡s  cCs5|jdƒx!|jƒD]}tj|ƒqWdS(sRemove ALL stored messages.iN(R5RRR RS(RRJ((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pytdelete_all_messagesªs cCs||j|jt|jdttƒƒD] \}}||kr)|Vq)q)WdS(s,Walk the files which are definitely pending.RON(R3t enumerateRRRERD(RR2RcRJ((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR:ÿs ) ccs‹|rt|ƒ}n|jƒ}xc|D][}xR|j|ƒD]A}t|j|ƒƒ}| sk||@ r>|j||ƒVq>q>Wq(WdS(N(RRuR]R (RRORyRRJRd((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRRs  tcCsQgtj|j|ƒƒD]}|jdƒs|^q}|jdd„ƒ|S(Ns.tmptkeycSst|jdƒdƒS(NRti(RwRT(R8((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyts(R RUR tendswithtsort(RtdirR8t message_files((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRuscGstjj|j|ŒS(N(R RRxR(Rtargs((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR scCs,t|ƒ}z|jƒSWd|jƒXdS(N(RktreadRm(RRJRq((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR>s c CsDd}|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. iiRN(R3R%RRR]RR?R>R@RARBRERjR RnRoR( RtoffsetR2RHt old_filenameRdRLRMtacceptedt new_filename((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR! s(      &#cCs3tjj|ƒ}d|kr/|jdƒdSdS(NRtiR}(R RtbasenameRT(RRRŠ((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyR]<s cCsztjj|ƒ\}}tjj||jdƒdƒ}|rf|ddjtt|ƒƒƒ7}ntj||ƒ|S(NRtiR}(R RRTRxR RRn(RRRdtdirnameRŠtnew_path((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRoBs "&cCs!|j||j|ƒ|ƒdS(N(RoR](RRRd((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRCJsiN(&t__name__t __module__t__doc__RRfttimeRRR#R%R&R(R*R,R-R/R1R3R5R6R;R<RNRYRZR\ReRrRjR:RRRuR R>R!R]RoRC(((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyRsD                  #        cOsGddlm}t||Ž}x!|jƒD]}|j|ƒq,W|S(sFGet 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_storeNs (RRRPRAR t landscape.libRt landscapeRRERDtobjectRR–(((s:/usr/lib/python2.7/dist-packages/landscape/broker/store.pyts    ÿ?