ó O'—^c@@sÐdZddlmZddlmZmZddlZddlZddlZddl Z ddl m Z ddl m Z ddlmZmZdd lmZdd lmZmZdZd jeeeƒƒZejd krBddlZddlZddlZejZ dZ!ej"Z#ej$ƒZ%d „Z&d„Z'nXejdkrŽddl(Z(ddl(m Z m!Z!m#Z#d„Z&d„Z'n e)dƒ‚de j*fd„ƒYZ*de j+fd„ƒYZ+dS(so kombu.transport.filesystem ========================== Transport using the file system as the message store. i(tabsolute_import(tloadstdumpsNi(tvirtual(t ChannelError(tEmptyt monotonic(tcached_property(t bytes_to_strt str_to_bytest.tntcC@s2tj|jƒƒ}tj||ddtƒdS(NiIÿÿ(t win32filet_get_osfhandletfilenot LockFileExt __overlapped(tfiletflagsthfile((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pytlock'scC@s/tj|jƒƒ}tj|ddtƒdS(NiIÿÿ(R R Rt UnlockFileExR(RR((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pytunlock+stposix(tLOCK_EXtLOCK_SHtLOCK_NBcC@stj|jƒ|ƒdS(N(tfcntltflockR(RR((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR4scC@stj|jƒtjƒdS(N(RRRtLOCK_UN(R((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR7ss9Filesystem plugin only defined for NT and POSIX platformstChannelcB@sweZd„Zd„Zd„Zd„Zed„ƒZed„ƒZ ed„ƒZ ed„ƒZ ed„ƒZ RS( cK@sÌdtttƒdƒƒtjƒ|f}tjj|j|ƒ}zly9t |dƒ}t |t ƒ|j t t|ƒƒƒWn,ttfk r®tdj|ƒƒ‚nXWdt|ƒ|jƒXdS(sPut `message` onto `queue`.s %s_%s.%s.msgiètwbs"Cannot add file {0!r} to directoryN(tinttroundRtuuidtuuid4tostpathtjointdata_folder_outtopenRRtwriteR RtIOErrortOSErrorRtformatRtclose(tselftqueuetpayloadtkwargstfilenametf((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyt_put@s  cC@sjd|d}tj|jƒ}t|ƒ}x.t|ƒdkr\|jdƒ}|j|ƒdkrkq/n|jr€|j}n t j ƒ}y&t j tj j|j|ƒ|ƒWntk rÅnXtj j||ƒ}yBt|dƒ}|jƒ}|jƒ|jstj|ƒnWn,ttfk rKtdj|ƒƒ‚nXtt|ƒƒSWtƒ‚dS(sGet next message from `queue`.R s.msgitrbs"Cannot read file {0!r} from queue.N(R$tlistdirtdata_folder_intsortedtlentpoptfindtstore_processedtprocessed_folderttempfilet gettempdirtshutiltmoveR%R&R*R(treadR-tremoveR+RR,RRR(R.R/t queue_findtfolderR2R=R3R0((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyt_getRs6         cC@s±d}d|d}tj|jƒ}x„t|ƒdkr¬|jƒ}yN|j|ƒdkrew)ntjj|j|ƒ}tj|ƒ|d7}Wq)t k r¨q)Xq)W|S(s!Remove all messages from `queue`.iR s.msgi( R$R6R7R9R:R;R%R&RCR+(R.R/tcountRDRER2((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyt_purgezs   cC@sud}dj|ƒ}tj|jƒ}xGt|ƒdkrp|jƒ}|j|ƒdkrcq*n|d7}q*W|S(s<Return the number of messages in `queue` as an :class:`int`.is.{0}.msgi(R,R$R6R7R9R:R;(R.R/RGRDRER2((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyt_size“s cC@s |jjjS(N(t connectiontclientttransport_options(R.((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyRL¤scC@s|jjddƒS(NR7tdata_in(RLtget(R.((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR7¨scC@s|jjddƒS(NR'tdata_out(RLRN(R.((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR'¬scC@s|jjdtƒS(NR<(RLRNtFalse(R.((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR<°scC@s|jjddƒS(NR=t processed(RLRN(R.((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR=´s( t__name__t __module__R4RFRHRItpropertyRLRR7R'R<R=(((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyR>s  (  t TransportcB@s)eZeZdZdZdZd„ZRS(it filesystemcC@sdS(NsN/A((R.((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pytdriver_versionÀs(RRRSRt default_portt driver_typet driver_nameRW(((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyRU¹s (iii(,t__doc__t __future__RtanyjsonRRR$R@R"R>tRtkombu.exceptionsRt kombu.fiveRRt kombu.utilsRtkombu.utils.encodingRR tVERSIONR&tmaptstrt __version__tnametwin32conR t pywintypestLOCKFILE_EXCLUSIVE_LOCKRRtLOCKFILE_FAIL_IMMEDIATELYRt OVERLAPPEDRRRRt RuntimeErrorRRU(((s</tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/filesystem.pyts@                {