mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Simpler threadless tarball streaming as suggested by James
This commit is contained in:
@@ -1,50 +1,26 @@
|
||||
"""
|
||||
A simple wrapper for writing tarballs as a stream. The work is performed in a
|
||||
thread and data is written to a Queue instead of a file.
|
||||
A simple wrapper for writing tarballs as a stream.
|
||||
"""
|
||||
|
||||
import logging, tarfile
|
||||
|
||||
from Queue import Queue, Empty, Full
|
||||
from threading import Thread
|
||||
|
||||
log = logging.getLogger( __name__ )
|
||||
|
||||
class QueueArchive( object ):
|
||||
queue_size = 32
|
||||
def __init__( self ):
|
||||
self.queue = Queue( QueueArchive.queue_size )
|
||||
self.get = self.queue.get
|
||||
self.empty = self.queue.empty
|
||||
def write( self, data ):
|
||||
self.queue.put( data, block=True, timeout=300 )
|
||||
def tell( self ):
|
||||
return 0
|
||||
|
||||
class StreamBall( object ):
|
||||
def __init__( self, mode, members={} ):
|
||||
self.mode = mode
|
||||
self.members = members
|
||||
self.tarfileobj = QueueArchive()
|
||||
self.wsgi_status = None
|
||||
self.wsgi_headeritems = None
|
||||
def add( self, file, relpath ):
|
||||
self.members[file] = relpath
|
||||
def stream( self ):
|
||||
t = Thread( target=self.thread_write )
|
||||
t.start()
|
||||
while t.isAlive():
|
||||
try:
|
||||
yield self.tarfileobj.get( block=False )
|
||||
except Empty:
|
||||
pass
|
||||
t.join()
|
||||
# exhaust the queue
|
||||
while not self.tarfileobj.empty():
|
||||
yield self.tarfileobj.get()
|
||||
def thread_write( self ):
|
||||
tf = tarfile.open( mode=self.mode, fileobj=self.tarfileobj )
|
||||
try:
|
||||
for file, rel in self.members.items():
|
||||
tf.add( file, arcname=rel )
|
||||
tf.close()
|
||||
except Full:
|
||||
log.warning( 'Queue full for longer than 300 seconds, timing out' )
|
||||
def stream( self, environ, start_response ):
|
||||
response_write = start_response( self.wsgi_status, self.wsgi_headeritems )
|
||||
class tarfileobj:
|
||||
def write( self, *args, **kwargs ):
|
||||
response_write( *args, **kwargs )
|
||||
tf = tarfile.open( mode=self.mode, fileobj=tarfileobj() )
|
||||
for file, rel in self.members.items():
|
||||
tf.add( file, arcname=rel )
|
||||
tf.close()
|
||||
return []
|
||||
|
||||
@@ -304,7 +304,9 @@ class Library( BaseController ):
|
||||
else:
|
||||
trans.response.set_content_type( "application/x-tar" )
|
||||
trans.response.headers[ "Content-Disposition" ] = "attachment; filename=GalaxyLibraryFiles.%s" % params.do_action
|
||||
return archive.stream()
|
||||
archive.wsgi_status = trans.response.wsgi_status()
|
||||
archive.wsgi_headeritems = trans.response.wsgi_headeritems()
|
||||
return archive.stream
|
||||
@web.expose
|
||||
def download_dataset_from_folder(self, trans, id, library_id=None, **kwd):
|
||||
"""Catches the dataset id and displays file contents as directed"""
|
||||
|
||||
Reference in New Issue
Block a user