def content(self):
        """
        Holt den Content aus dem *_content*-Attribut oder aus dem
        *_blobs*-Ordner und gibt diesen entpackt zurück.
        """

        # Falls der Content noch nicht gespeichert wurde, befindet
        # er sich noch in *self._content*.
        if not self._content is None:
            return self._content

        # Nachsehen ob es einen Blob für den Content gibt
        if not self.content_blob_name:
            return None

        # Blob laden, entpacken und zurück geben
        blob_dir = os.path.join(config.DATABLOBSDIR.value, self.content_blob_name[0])
        blob_path = os.path.join(blob_dir, self.content_blob_name)
        with io.open(blob_path, "rb") as blob_file:
            if self.content_blob_name.endswith(".snappy"):
                content = snappy.uncompress(blob_file.read())
            else:
                content = blob_file.read()
            if content and self.node.content_type in constants.CONTENT_TYPES_TEXT:
                return content.decode("utf-8")
            else:
                return content

    def take_action(self, parsed_args):
        self.log = logging.getLogger(self.__class__.__name__)
        self.log.debug('Initialized %s', self.__class__.__name__)
        self.log.debug('Args: %s', str(parsed_args))

        self.log.debug("Establishing connection")

        c, db, collection = OutputDBInterface.get_db_connection(hostname=parsed_args.hostname)

        cursor = collection.find()
        N = cursor.count()
        if N == 0:
            self.log.error("No events in the output database; no file made.")
            return

        f = gzip.open(parsed_args.filename, 'wb')

        pickle.dump(__version__, f)

        self.log.info("Processing %d trigger events" % N)

        for i in tqdm(range(N)):
            doc = next(cursor)
            doc2 = snappy.uncompress(doc['compressed_doc'])
            doc2 = pickle.loads(doc2)

            pickle.dump(doc2, f)


        f.close()

def decompress(data):
    """
    Decompresses the given data via the snappy algorithm.

    If ``python-snappy`` is not installed a ``RuntimeError`` is raised.
    """
    if not snappy_available:
        raise RuntimeError("Snappy compression unavailable.")

    buff_offset = len(raw_header)  # skip the header
    length = len(data) - len(raw_header)

    output = BytesIO()

    while buff_offset <= length:
        block_size = struct.unpack_from("!i", data, buff_offset)[0]
        buff_offset += struct.calcsize("!i")

        block = struct.unpack_from("!%ds" % block_size, data, buff_offset)[0]
        buff_offset += block_size

        output.write(snappy.uncompress(block))

    result = output.getvalue()

    output.close()

    return result

    def __init__(self,  filename):
        self.fileName = filename
        self.api = "API_UNKNOWN"
        self.traceFile = open(self.fileName, 'rb+')
        self.filePointer = 0
        self.fileSize = os.path.getsize(self.fileName)
        self.nextCallNumber = 0
        self.lastFrameBreakPos = 0

        self.container = 0
        self.containerPointer = 0

        self.fullFilePosition = 0

        self.mem = self.traceFile.read(2)
        self.filePointer += 2

        if self.mem[0:2] != b'at':
            raise Exception("not snappy file!")

        length = int(struct.unpack('I', self.traceFile.read(4))[0])
        self.filePointer += 4

        compressedMem = self.traceFile.read(length)
        self.filePointer += length

        self.mem = snappy.uncompress(compressedMem)
        self.getVersion(self.mem)
        return

def print_journal(fn, at=0):

    # file structure
    header_struct = xstruct('< 5s 20s 1s 128s 2s Q')
    section_struct = xstruct('< I Q Q')
    footer_struct = xstruct('< I QQ Q 4s')
    align = 8192

    # open file
    f = open(fn, 'rb')
    sz = os.fstat(f.fileno()).st_size # 2.4 won't accept 0
    buf = mmap.mmap(f.fileno(), sz, prot=mmap.PROT_READ)
    
    # file header
    magic, date, _, path, _, fileid = unpack_from(header_struct, buf)
    path = path[:path.find('\0')]
    date = date[:date.find('\0')]
    print '%08x: header magic=%s date=%s path=%s fid=%x' % (0, hex(magic), date, path, fileid)
    if at==0:
        at = 8192
    
    # traverse file
    while at < len(buf):
    
        # section header
        l, lsn, fid = unpack_from(section_struct, buf, at)
        lp = (l + align-1) & ~(align-1)
        section_at = at + 20
        footer_at = at + l - 32
        ok = 'OK'
        if fid!=fileid: ok = 'BAD'
        print '%08x: section l=%x(%d) lp=%x(%d) lsn=%x(%d) fid=%x(%s)' % \
            (at, l, l, lp, lp, lsn, lsn, fid, ok)
    
        # compute hash, compare with footer
        sentinel, hash_a, hash_b, reserved, magic = unpack_from(footer_struct, buf, footer_at)
        computed_hash_a, computed_hash_b = hash(buf[at:footer_at])
        hash_ok = 'OK'
        if not (hash_a==computed_hash_a and hash_b==computed_hash_b): ok = 'BAD'
        print '%08x: hash=%08x:%08x(%s)' % (at, computed_hash_a, computed_hash_b, hash_ok)
    
        # section
        try:
            if snappy:
                section = snappy.uncompress(buf[section_at:footer_at])
                print '%08x: uncompressed length=%x(%d)' % (section_at, len(section), len(section))
                if do_journal_entries:
                    print_journal_entries(section)
        except Exception, e:
            print '%08x: %s' % (section_at, e)
    
        # section footer
        print '%08x: footer sentinel=%x hash=%08x:%08x(%s) magic=%s' % \
            (footer_at, sentinel, hash_a, hash_b, hash_ok, hex(magic))
    
        # next section
        at += lp

def _unpack_msgpack_snappy(str):
    if str.startswith(b'S'):
        tmp = snappy.uncompress(str[1:])
        # print "SNAPPY: ", len(str), len(tmp)
        obj = msgpack.loads(tmp, encoding='utf-8')
    elif str.startswith(b'\0'):
        obj = msgpack.loads(str[1:], encoding='utf-8')
    else:
        return None
    
    return obj

def _unpack(str) :

    if str[0] == 'S':
        tmp = snappy.uncompress(str[1:])
        obj = msgpack.loads(tmp)
    elif str[0] == '\0':
        obj = msgpack.loads(str[1:])
    else:
        return None

    #print "UNPACK", obj
    return obj

 def getByte(self):
     if self.containerPointer == len(self.mem):
         length = int(struct.unpack('I', self.traceFile.read(4))[0])
         self.filePointer += 4
         compressedMem = self.traceFile.read(length)
         self.filePointer += length
         self.container += 1
         self.mem = snappy.uncompress(compressedMem)
         self.containerPointer = 0
     rval= self.mem[self.containerPointer]
     self.containerPointer += 1
     self.fullFilePosition += 1
     return rval

def decompress(data, compressor_id):
    if compressor_id == SnappyContext.compressor_id:
        # python-snappy doesn't support the buffer interface.
        # https://github.com/andrix/python-snappy/issues/65
        # This only matters when data is a memoryview since
        # id(bytes(data)) == id(data) when data is a bytes.
        # NOTE: bytes(memoryview) returns the memoryview repr
        # in Python 2.7. The right thing to do in 2.7 is call
        # memoryview.tobytes(), but we currently only use
        # memoryview in Python 3.x.
        return snappy.uncompress(bytes(data))
    elif compressor_id == ZlibContext.compressor_id:
        return zlib.decompress(data)
    else:
        raise ValueError("Unknown compressorId %d" % (compressor_id,))

 def _decompress_subblock(self):
     if self._subblock_size is None:
         if len(self._buf) <= 4:
             return b""
         self._subblock_size = struct.unpack(">i", self._buf[:4])[0]
         self._buf = self._buf[4:]
     # Only attempt to decompress complete subblocks.
     if len(self._buf) < self._subblock_size:
         return b""
     compressed = self._buf[:self._subblock_size]
     self._buf = self._buf[self._subblock_size:]
     uncompressed = snappy.uncompress(compressed)
     self._block_read += len(uncompressed)
     self._subblock_size = None
     return uncompressed

def read(stream, peek):
  if peek.cls not in PBMSG_BY_KIND.values():
    msg = 'please update demo.proto: {0}'.format(peek.cls)
    raise InvalidProtobufMessage(msg)

  stream.seek(peek.offset)

  data = stream.read(peek.size)
  if peek.compressed:
    data = snappy.uncompress(data)

  message = peek.cls()
  message.ParseFromString(data)

  return message

    def post(self):
        """Insert a message
        """
        # print self.request.body
        print "get message"
        userid = self.get_argument("userid")
        pcid = self.get_argument("pcid")
        packettype = self.get_argument("type")
        pos = self.request.body.find("&data=")

        if pos==-1:
            self.write('error')
            self.finish()
            return

        snappydata = self.request.body[pos+6:]

        try:
            protodata=snappy.uncompress(snappydata)
            for package in protodata.split("!!!"):

                if len(package) < 10:
                    continue
                if packettype == "1":
                    packetobj = IpPacket()
                    packetobj.ParseFromString(package)
                    self.saveipdata(packetobj,userid,pcid)
                elif packettype == "2":
                    packetobj = EmailPacket()
                    packetobj.ParseFromString(package)
                    self.saveemaildata(packetobj,userid,pcid)
                elif packettype == "3":
                    packetobj = HttpPacket()
                    packetobj.ParseFromString(package)
                    self.savehttpdata(packetobj,userid,pcid)
        except:
            print "error"
            print "print self.request.body",self.request.body
            print "self.request.arguments",self.request.arguments
            traceback.print_exc()
            self.write('ok')
            self.finish()
            return
            
        self.write('ok')
        self.finish()

def imgmsg_to_pil(img_msg, rgba=True):
    try:
        uncompressed_img_msg = sensor_msgs.msg.Image()
        uncompressed_img_msg.header = img_msg.header
        uncompressed_img_msg.height = img_msg.height
        uncompressed_img_msg.width = img_msg.width
        uncompressed_img_msg.step = 1
        uncompressed_img_msg.encoding = 'mono8'
        uncompressed_img_msg.data = snappy.uncompress(np.fromstring(img_msg.data, dtype = 'uint8'))
        if img_msg._type == 'sensor_msgs/CompressedImage':
            pil_img = Image.open(StringIO(img_msg.data))
            if pil_img.mode != 'L':
                pil_img = pil_bgr2rgb(pil_img)
        else:
            alpha = False
            if uncompressed_img_msg.encoding == 'mono8':
                mode = 'L'
            elif uncompressed_img_msg.encoding == 'rgb8':
                mode = 'BGR'
            elif uncompressed_img_msg.encoding == 'bgr8':
                mode = 'RGB'
            elif uncompressed_img_msg.encoding in ['bayer_rggb8', 'bayer_bggr8', 'bayer_gbrg8', 'bayer_grbg8']:
                mode = 'L'
            elif uncompressed_img_msg.encoding == 'mono16':
                if uncompressed_img_msg.is_bigendian:
                    mode = 'F;16B'
                else:
                    mode = 'F:16'
            elif uncompressed_img_msg.encoding == 'rgba8':
                mode = 'BGR'
                alpha = True
            elif uncompressed_img_msg.encoding == 'bgra8':
                mode = 'RGB'
                alpha = True
    
            pil_img = Image.frombuffer('RGB', (uncompressed_img_msg.width, uncompressed_img_msg.height), uncompressed_img_msg.data, 'raw', mode, 0, 1)

        if rgba and pil_img.mode != 'RGBA':
            pil_img = pil_img.convert('RGBA')
    
        return pil_img

    except Exception, ex:
        print >> sys.stderr, 'Can\'t convert image: %s' % ex
        return None

    def process_raw(self, ctx, evidence_uuid, pipeline, data, raw, return_result=False, autosave=True):
        pipeline = self._get_pipeline(pipeline)
        if not pipeline:
            return

        # StringIO for raw data
        stream = StringIO.StringIO(snappy.uncompress(base64.b64decode(raw)))

        # Perform the actual processing
        bundle = Bundle(self.server, evidence_uuid, pipeline, data, stream)
        try:
            bundle = pipeline.process(bundle)
        except Exception as e:
            bundle.add_exception(e, traceback=''.join(traceback.format_exc()))

        data = bundle.data
        if autosave:
            data = self._save_result(evidence_uuid, data, wait=return_result)

        if return_result:
            return data

def imgmsg_to_pil(img_msg, rgba=True):
    try:
        uncompressed_img_msg = sensor_msgs.msg.Image()
        uncompressed_img_msg.header = img_msg.header
        uncompressed_img_msg.height = img_msg.height
        uncompressed_img_msg.width = img_msg.width
        uncompressed_img_msg.step = 1
        uncompressed_img_msg.encoding = 'mono8'
        uncompressed_img_msg.data = snappy.uncompress(np.fromstring(img_msg.data, dtype = 'uint8'))
        alpha = False
        mode = 'L'
        pil_img = Image.frombuffer('RGB', (uncompressed_img_msg.width, uncompressed_img_msg.height), uncompressed_img_msg.data, 'raw', mode, 0, 1)

        if rgba and pil_img.mode != 'RGBA':
            pil_img = pil_img.convert('RGBA')

        return pil_img

    except Exception, ex:
        print >> sys.stderr, 'Can\'t convert image: %s' % ex
        return None

    def items(self, filter=None):
        '''Yield a pair of (name, data) records in the same
        order they appear in the file.
        @p filter -- same as for get function
        '''
        records = {}

        for block_type, block_data in self._read_blocks():
            if block_type == _BLOCK_SCHEMA:
                identifier, bulk_record = self._parse_schema(
                    block_data, filter=filter)
                if identifier is None:
                    continue
                records[identifier] = bulk_record
            elif block_type == _BLOCK_DATA:
                stream = telemetry_archive.ReadStream(
                    stringio.StringIO(block_data))
                identifier = stream.read_uint32()
                flags = stream.read_uint16()

                record = records.get(identifier, None)
                if record is None:
                    continue

                if flags & BlockDataFlags.kPreviousOffset:
                    flags &= ~(BlockDataFlags.kPreviousOffset)
                    _ = stream.read_varint()
                if flags & BlockDataFlags.kSchemaCRC:
                    assert False  # not supported yet
                if flags & BlockDataFlags.kSnappy:
                    flags &= ~(BlockDataFlags.kSnappy)
                    rest = stream.stream.read()
                    stream = telemetry_archive.ReadStream(
                        stringio.StringIO(snappy.uncompress(rest)))
                assert flags == 0  # no unknown flags

                rest = stream.stream.read()
                yield record.name, record.deserialize(rest)

def get_samples_from_doc(doc, is_compressed):
    """From a mongo document, fetch the data payload and decompress if
    necessary

    Args:
       doc (dictionary):  Document from mongodb to analyze

    Returns:
       bytes: decompressed data

    """
    data = doc['data']
    assert len(data) != 0

    if is_compressed:
        data = snappy.uncompress(data)

    data = np.fromstring(data,
                         dtype=SAMPLE_TYPE)
    if len(data) == 0:
        raise IndexError("Data has zero length")

    return data

def get_data_from_doc(doc):
    """From a mongo document, fetch the data payload and decompress if
    necessary

    Args:
       doc (dictionary):  Document from mongodb to analyze

    Returns:
       bytes: decompressed data

    """
    data = doc['data']
    assert len(data) != 0

    if doc['zipped']:
        data = snappy.uncompress(data)

    data = np.fromstring(data,
                         dtype=np.uint32)

    if len(data) == 0:
        raise IndexError("Data has zero length")

    return data

 def getImageFeatures(self, image , image_shape):
     image = np.fromstring(snappy.uncompress(image), dtype=np.float32)
     image.resize(image_shape)
     feature_dic = self.extractor.getImageFeatures(image)
     feature_dic = {layer:snappy.compress(features) for layer,features in feature_dic.items()}
     return feature_dic

#.........这里部分代码省略.........
    # buffer position so that we can return to it if the request fails
    # and needs to be retried.
    rewind_input_buffer_offset = None
    if hasattr(data, 'seek') and hasattr(data, 'tell'):
        rewind_input_buffer_offset = data.tell()

    headers['DNAnexus-API'] = API_VERSION
    headers['User-Agent'] = USER_AGENT

    if use_compression == 'snappy':
        if not snappy_available:
            raise DXError("Snappy compression requested, but the snappy module is unavailable")
        headers['accept-encoding'] = 'snappy'

    if 'verify' not in kwargs and 'DX_CA_CERT' in os.environ:
        kwargs['verify'] = os.environ['DX_CA_CERT']
        if os.environ['DX_CA_CERT'] == 'NOVERIFY':
            kwargs['verify'] = False

    response, last_error = None, None
    for retry in range(max_retries + 1):
        streaming_response_truncated = False
        try:
            response = session_handler.request(method, url, data=data, headers=headers, timeout=timeout, auth=auth,
                                               **kwargs)

            if _UPGRADE_NOTIFY and response.headers.get('x-upgrade-info', '').startswith('A recommended update is available') and not os.environ.has_key('_ARGCOMPLETE'):
                logger.info(response.headers['x-upgrade-info'])
                try:
                    with file(_UPGRADE_NOTIFY, 'a'):
                        os.utime(_UPGRADE_NOTIFY, None)
                except:
                    pass
                _UPGRADE_NOTIFY = False

            if _DEBUG:
                print >>sys.stderr, method, url, "<=", response.status_code, Repr().repr(response.content)

            # If HTTP code that is not 200 (OK) is received and the content is
            # JSON, parse it and throw the appropriate error.  Otherwise,
            # raise the usual exception.
            if response.status_code != requests.codes.ok:
                # response.headers key lookup is case-insensitive
                if response.headers.get('content-type', '').startswith('application/json'):
                    content = json.loads(response.content)
                    raise DXAPIError(content,
                                     response.status_code)
                response.raise_for_status()

            if want_full_response:
                return response
            else:
                if 'content-length' in response.headers:
                    if int(response.headers['content-length']) != len(response.content):
                        raise ContentLengthError("Received response with content-length header set to %s but content length is %d"
                            % (response.headers['content-length'], len(response.content)))

                if use_compression and response.headers.get('content-encoding', '') == 'snappy':
                    # TODO: check if snappy raises any exceptions on truncated response content
                    decoded_content = snappy.uncompress(response.content)
                else:
                    decoded_content = response.content

                if response.headers.get('content-type', '').startswith('application/json'):
                    try:
                        return json.loads(decoded_content)
                    except ValueError:
                        # If a streaming API call (no content-length
                        # set) encounters an error it may just halt the
                        # response because it has no other way to
                        # indicate an error. Under these circumstances
                        # the client sees unparseable JSON, and we
                        # should be able to recover.
                        streaming_response_truncated = 'content-length' not in response.headers
                        raise HTTPError("Invalid JSON received from server")
                return decoded_content
        except (DXAPIError, ConnectionError, HTTPError, Timeout, httplib.HTTPException) as e:
            last_error = e

            # TODO: support HTTP/1.1 503 Retry-After
            # TODO: if the socket was dropped mid-request, ConnectionError or httplib.IncompleteRead is raised,
            # but non-idempotent requests can be unsafe to retry
            # Distinguish between connection initiation errors and dropped socket errors
            if retry < max_retries:
                if (response is None) or isinstance(e, ContentLengthError):
                    ok_to_retry = always_retry or (method == 'GET')
                else:
                    ok_to_retry = (response.status_code >= 500 and response.status_code < 600) or streaming_response_truncated

                if ok_to_retry:
                    if rewind_input_buffer_offset is not None:
                        data.seek(rewind_input_buffer_offset)
                    delay = 2 ** (retry+1)
                    logger.warn("%s %s: %s. Waiting %d seconds before retry %d of %d..." % (method, url, str(e), delay, retry+1, max_retries))
                    time.sleep(delay)
                    continue
            break
        if last_error is None:
            last_error = DXError("Internal error in DXHTTPRequest")
    raise last_error