import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private final BufferedImage retrieveLogoImage()
{
	return logoCache.get(0, new Callable<Optional<BufferedImage>>()
	{

		@Override
		public Optional<BufferedImage> call() throws Exception
		{
			final I_AD_Image logo = retrieveLogoADImage();
			if (logo == null)
			{
				return Optional.absent();
			}

			final byte[] data = logo.getBinaryData();
			if (data == null)
			{
				return Optional.absent();
			}

			final BufferedImage logoImage = ImageIO.read(new ByteArrayInputStream(data));
			return Optional.fromNullable(logoImage);
		}
	}).orNull();
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
/**
 * Builds the message body from data
 *
 * @throws JMSException
 * @throws IOException
 */
private void loadContent() throws JMSException {
    try {
        if (getContent() != null && map.isEmpty()) {
            ByteSequence content = getContent();
            InputStream is = new ByteArrayInputStream(content);
            if (isCompressed()) {
                is = new InflaterInputStream(is);
            }
            DataInputStream dataIn = new DataInputStream(is);
            map = MarshallingSupport.unmarshalPrimitiveMap(dataIn);
            dataIn.close();
        }
    } catch (IOException e) {
        throw JMSExceptionSupport.create(e);
    }
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
/**
 * Gets the serializable object containing this message's data. The default
 * value is null.
 *
 * @return the serializable object containing this message's data
 * @throws JMSException
 */
public Serializable getObject() throws JMSException {
    if (object == null && getContent() != null) {
        try {
            ByteSequence content = getContent();
            InputStream is = new ByteArrayInputStream(content);
            if (isCompressed()) {
                is = new InflaterInputStream(is);
            }
            DataInputStream dataIn = new DataInputStream(is);
            ClassLoadingAwareObjectInputStream objIn = new ClassLoadingAwareObjectInputStream(dataIn);
            try {
                object = (Serializable)objIn.readObject();
            } catch (ClassNotFoundException ce) {
                throw JMSExceptionSupport.create("Failed to build body from content. Serializable class not available to broker. Reason: " + ce, ce);
            } finally {
                dataIn.close();
            }
        } catch (IOException e) {
            throw JMSExceptionSupport.create("Failed to build body from bytes. Reason: " + e, e);
        }
    }
    return this.object;
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private boolean evaluate(byte[] data) {
    try {

        InputSource inputSource = new InputSource(new ByteArrayInputStream(data));

        DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
        factory.setNamespaceAware(true);
        DocumentBuilder dbuilder = factory.newDocumentBuilder();
        Document doc = dbuilder.parse(inputSource);
        
        CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
        XObject result = cachedXPathAPI.eval(doc, xpath);
        if (result.bool())
        	return true;
        else {
        	NodeIterator iterator = cachedXPathAPI.selectNodeIterator(doc, xpath);
        	return (iterator.nextNode() != null);
        }  

    } catch (Throwable e) {
        return false;
    }
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private HttpResponse response(final int status, String value) throws IOException {
  HttpResponse httpResponse = mock(HttpResponse.class);
  when(httpResponse.getStatusLine()).thenReturn(new StatusLine() {
    @Override
    public ProtocolVersion getProtocolVersion() {
      return new ProtocolVersion("http", 1, 1);
    }

    @Override
    public int getStatusCode() {
      return status;
    }

    @Override
    public String getReasonPhrase() {
      return "";
    }
  });
  when(httpResponse.getAllHeaders()).thenReturn(new Header[]{});
  HttpEntity entity = mock(HttpEntity.class);
  when(entity.getContent()).thenReturn(new ByteArrayInputStream(value.getBytes()));
  when(httpResponse.getEntity()).thenReturn(entity);
  return httpResponse;
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
/**
 * Generates ARCLib XML from SIP using the SIP profile
 *
 * @param sipPath      path to the SIP package
 * @param sipProfileId id of the SIP profile
 * @return ARCLib XML
 * @throws IOException
 * @throws SAXException
 * @throws ParserConfigurationException
 * @throws TransformerException
 */
public String generateArclibXml(String sipPath, String sipProfileId)
        throws ParserConfigurationException, TransformerException, SAXException, IOException {
    log.info("Generating ARCLib XML for SIP at path " + sipPath + " using SIP profile with ID " + sipProfileId + ".");

    SipProfile sipProfile = store.find(sipProfileId);
    notNull(sipProfile, () -> new MissingObject(SipProfile.class, sipProfileId));

    ArclibXmlValidator.validateWithXMLSchema(
            new ByteArrayInputStream(sipProfile.getXml().getBytes(StandardCharsets.UTF_8.name())),
            new InputStream[]{sipProfileSchema.getInputStream()});

    String sipProfileXml = sipProfile.getXml();
    notNull(sipProfileXml, () -> new InvalidAttribute(sipProfile, "xml", null));

    Document arclibXmlDoc = DocumentHelper.createDocument();
    arclibXmlDoc.addElement(new QName(ROOT, Namespace.get("METS", uris.get("METS"))));

    NodeList mappingNodes = XPathUtils.findWithXPath(stringToInputStream(sipProfileXml), MAPPING_ELEMENTS_XPATH);
    for (int i = 0; i < mappingNodes.getLength(); i++) {
        Set<Utils.Pair<String, String>> nodesToCreate = nodesToCreateByMapping((Element) mappingNodes.item(i), sipPath);

        XmlBuilder xmlBuilder = new XmlBuilder(uris);
        for (Utils.Pair<String, String> xPathToValue : nodesToCreate) {
            xmlBuilder.addNode(arclibXmlDoc, xPathToValue.getL(), xPathToValue.getR(), uris.get("ARCLIB"));
        }
    }

    String arclibXml = arclibXmlDoc.asXML().replace("&lt;", "<").replace("&gt;", ">");
    log.info("Generated ARCLib XLM: \n" + arclibXml);

    return arclibXml;
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
public Command read() throws IOException {
    Command answer = null;
    Endpoint from = null;
    synchronized (readLock) {
        while (true) {
            DatagramPacket datagram = createDatagramPacket();
            channel.receive(datagram);

            // TODO could use a DataInput implementation that talks direct
            // to the byte[] to avoid object allocation
            receiveCounter++;
            DataInputStream dataIn = new DataInputStream(new ByteArrayInputStream(datagram.getData(), 0, datagram.getLength()));
            
            from = headerMarshaller.createEndpoint(datagram, dataIn);
            answer = (Command)wireFormat.unmarshal(dataIn);
            break;
        }
    }
    if (answer != null) {
        answer.setFrom(from);

        if (LOG.isDebugEnabled()) {
            LOG.debug("Channel: " + name + " about to process: " + answer);
        }
    }
    return answer;
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
public Command read() throws IOException {
    Command answer = null;
    Endpoint from = null;
    synchronized (readLock) {
        while (true) {
            readBuffer.clear();
            SocketAddress address = channel.receive(readBuffer);

            readBuffer.flip();

            if (readBuffer.limit() == 0) {
                continue;
            }
            
            receiveCounter++;
            from = headerMarshaller.createEndpoint(readBuffer, address);

            int remaining = readBuffer.remaining();
            byte[] data = new byte[remaining];
            readBuffer.get(data);

            // TODO could use a DataInput implementation that talks direct
            // to
            // the ByteBuffer to avoid object allocation and unnecessary
            // buffering?
            DataInputStream dataIn = new DataInputStream(new ByteArrayInputStream(data));
            answer = (Command)wireFormat.unmarshal(dataIn);
            break;
        }
    }
    if (answer != null) {
        answer.setFrom(from);

        if (LOG.isDebugEnabled()) {
            LOG.debug("Channel: " + name + " received from: " + from + " about to process: " + answer);
        }
    }
    return answer;
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private boolean evaluate(byte[] data) {
    try {
        InputSource inputSource = new InputSource(new ByteArrayInputStream(data));
        return ((Boolean)expression.evaluate(inputSource, XPathConstants.BOOLEAN)).booleanValue();
    } catch (XPathExpressionException e) {
        return false;
    }
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
@Test
public void testLogFile() throws Exception {
	FileSystem fs = dfsCluster.getFileSystem();
	OutputStream out = fs.create(logFile);
	InputStream in = new ByteArrayInputStream(
			("instancePaths=/ivory/feed/agg-logs/path1/2010/10/10/20,"
					+ "/ivory/feed/agg-logs/path1/2010/10/10/21,"
					+ "/ivory/feed/agg-logs/path1/2010/10/10/22,"
					+ "/ivory/feed/agg-logs/path1/2010/10/10/23")
					.getBytes());
	IOUtils.copyBytes(in, out, conf);
	testProcessMessageCreator();
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
@Test
public void testEmptyLogFile() throws Exception {
	FileSystem fs = dfsCluster.getFileSystem();
	OutputStream out = fs.create(logFile);
	InputStream in = new ByteArrayInputStream(("instancePaths=").getBytes());
	IOUtils.copyBytes(in, out, conf);

	new MessageProducer().run(this.args);
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
@Test
public void testLogFile() throws Exception {
    FileSystem fs = dfsCluster.getFileSystem();
    OutputStream out = fs.create(logFile);
    InputStream in = new ByteArrayInputStream(
            ("instancePaths=/falcon/feed/agg-logs/path1/2010/10/10/20,"
                    + "/falcon/feed/agg-logs/path1/2010/10/10/21,"
                    + "/falcon/feed/agg-logs/path1/2010/10/10/22,"
                    + "/falcon/feed/agg-logs/path1/2010/10/10/23")
                    .getBytes());
    IOUtils.copyBytes(in, out, conf);
    testProcessMessageCreator();
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
@Test
public void testEmptyLogFile() throws Exception {
    FileSystem fs = dfsCluster.getFileSystem();
    OutputStream out = fs.create(logFile);
    InputStream in = new ByteArrayInputStream(("instancePaths=").getBytes());
    IOUtils.copyBytes(in, out, conf);

    new MessageProducer().run(this.args);
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private Source createXMLSourceFromString(final String xmlStr)
{
	final InputStream inputStream = new ByteArrayInputStream(xmlStr == null ? new byte[0] : xmlStr.getBytes());
	return new StreamSource(inputStream);
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
public byte[] getScaledImageData(final int maxWidth, final int maxHeight)
{
	if (maxWidth <= 0 && maxHeight <= 0)
	{
		return getData();
	}

	final BufferedImage image;
	try
	{
		image = ImageIO.read(new ByteArrayInputStream(getData()));
	}
	catch (IOException ex)
	{
		throw AdempiereException.wrapIfNeeded(ex);
	}
	if (image == null)
	{
		return null;
	}

	final Dimension sizeOrig = new Dimension(image.getWidth(), image.getHeight());
	final Dimension sizeNew = getScaledDimension(sizeOrig, new Dimension(maxWidth, maxHeight));
	if (sizeNew.equals(sizeOrig))
	{
		return getData();
	}
	else
	{
		final BufferedImage resizedImage = new BufferedImage(sizeNew.width, sizeNew.height, BufferedImage.TYPE_INT_RGB);
		final Graphics2D g = resizedImage.createGraphics();

		g.setComposite(AlphaComposite.Src);
		g.setRenderingHint(RenderingHints.KEY_INTERPOLATION, RenderingHints.VALUE_INTERPOLATION_BILINEAR);
		g.setRenderingHint(RenderingHints.KEY_RENDERING, RenderingHints.VALUE_RENDER_QUALITY);
		g.setRenderingHint(RenderingHints.KEY_ANTIALIASING, RenderingHints.VALUE_ANTIALIAS_ON);

		g.drawImage(image, 0, 0, sizeNew.width, sizeNew.height, null);
		g.dispose();

		return toByteArray(resizedImage, getImageFormatName());
	}
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
public Object unmarshal(ByteSequence packet) throws IOException {
    ByteArrayInputStream stream = new ByteArrayInputStream(packet);
    DataInputStream dis = new DataInputStream(stream);
    return unmarshal(dis);
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
protected HashMap<String, String> parseHeaders(DataInput in) throws IOException {
    HashMap<String, String> headers = new HashMap<String, String>(25);
    while (true) {
        ByteSequence line = readHeaderLine(in, MAX_HEADER_LENGTH, "The maximum header length was exceeded");
        if (line != null && line.length > 1) {

            if (headers.size() > MAX_HEADERS) {
                throw new ProtocolException("The maximum number of headers was exceeded", true);
            }

            try {

                ByteArrayInputStream headerLine = new ByteArrayInputStream(line);
                ByteArrayOutputStream stream = new ByteArrayOutputStream(line.length);

                // First complete the name
                int result = -1;
                while ((result = headerLine.read()) != -1) {
                    if (result != ':') {
                        stream.write(result);
                    } else {
                        break;
                    }
                }

                ByteSequence nameSeq = stream.toByteSequence();

                String name = new String(nameSeq.getData(), nameSeq.getOffset(), nameSeq.getLength(), "UTF-8");
                String value = decodeHeader(headerLine);
                if (stompVersion.equals(Stomp.V1_0)) {
                    value = value.trim();
                }

                if (!headers.containsKey(name)) {
                    headers.put(name, value);
                }

                stream.close();

            } catch (Exception e) {
                throw new ProtocolException("Unable to parser header line [" + line + "]", true);
            }
        } else {
            break;
        }
    }
    return headers;
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
public Object unmarshal(ByteSequence packet) throws IOException {
    return unmarshal(new DataInputStream(new ByteArrayInputStream(packet)));
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private Map<String, Object> unmarsallProperties(ByteSequence marshalledProperties) throws IOException {
    return MarshallingSupport.unmarshalPrimitiveMap(new DataInputStream(new ByteArrayInputStream(marshalledProperties)));
}
 

import org.apache.activemq.util.ByteArrayInputStream; //导入依赖的package包/类
private Map<String, Object> unmarsallProperties(ByteSequence marshalledProperties) throws IOException {
    return MarshallingSupport.unmarshalPrimitiveMap(new DataInputStream(new ByteArrayInputStream(marshalledProperties)), MAX_PROPERTY_SIZE);
}