You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ju...@apache.org on 2011/01/17 12:22:56 UTC
svn commit: r1059879 - in
/tika/trunk/tika-core/src/main/java/org/apache/tika/fork:
ClassLoaderResource.java ForkClient.java ForkResource.java ForkServer.java
Author: jukka
Date: Mon Jan 17 11:22:56 2011
New Revision: 1059879
URL: http://svn.apache.org/viewvc?rev=1059879&view=rev
Log:
TIKA-416: Out-of-process text extraction
Add a concept of fork resources, that the forked server process can use to access information from the parent process.
Added:
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderResource.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkResource.java
Modified:
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java
Added: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderResource.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderResource.java?rev=1059879&view=auto
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderResource.java (added)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderResource.java Mon Jan 17 11:22:56 2011
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.fork;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.Enumeration;
+
+class ClassLoaderResource implements ForkResource {
+
+ private final ClassLoader loader;
+
+ public ClassLoaderResource(ClassLoader loader) {
+ this.loader = loader;
+ }
+
+ /**
+ * Processes a request for one (code 1) or many (code 2) class loader
+ * resources. The requested resources are sent preceded with a boolean
+ * <code>true</code> value. If the resource was not found (code 1) or
+ * when the last resource has been sent (code 2), a boolean
+ * <code>false</code> value is sent instead.
+ *
+ * @param name resource name
+ * @throws IOException if the resource could not be sent
+ */
+ public void process(DataInputStream input, DataOutputStream output)
+ throws IOException {
+ byte type = input.readByte();
+ String name = input.readUTF();
+ System.out.println(name);
+ if (type == 1) {
+ InputStream stream = loader.getResourceAsStream(name);
+ if (stream != null) {
+ output.writeBoolean(true);
+ writeAndCloseStream(output, stream);
+ } else {
+ output.writeBoolean(false);
+ }
+ } else if (type == 2) {
+ Enumeration<URL> resources = loader.getResources(name);
+ while (resources.hasMoreElements()) {
+ output.writeBoolean(true);
+ InputStream stream = resources.nextElement().openStream();
+ writeAndCloseStream(output, stream);
+ }
+ output.writeBoolean(false);
+ }
+ output.flush();
+ }
+
+ /**
+ * Sends the contents of the given input stream to the given output.
+ * The stream is sent in chunks of less than 64kB, each preceded by
+ * a 16-bit integer value that indicates the length of the following
+ * chunk. A zero short value is sent at the end to signify the end of
+ * the stream.
+ * <p>
+ * The stream is guaranteed to be closed by this method, regardless of
+ * the way it returns.
+ *
+ * @param stream the stream to be sent
+ * @throws IOException if the stream could not be sent
+ */
+ private void writeAndCloseStream(
+ DataOutputStream output, InputStream stream) throws IOException {
+ try {
+ byte[] buffer = new byte[0x10000 - 1];
+ int n;
+ while ((n = stream.read(buffer)) != -1) {
+ System.out.println(n);
+ output.writeShort(n);
+ output.write(buffer, 0, n);
+ }
+ output.writeShort(0);
+ } finally {
+ stream.close();
+ }
+ }
+
+}
Modified: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java?rev=1059879&r1=1059878&r2=1059879&view=diff
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java (original)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java Mon Jan 17 11:22:56 2011
@@ -24,7 +24,9 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URL;
+import java.util.ArrayList;
import java.util.Enumeration;
+import java.util.List;
import org.apache.tika.io.IOExceptionWithCause;
import org.apache.tika.io.IOUtils;
@@ -82,6 +84,7 @@ class ForkClient {
*/
private void copyClassToDirectory(Class<?> klass) throws IOException {
String path = klass.getName().replace('.', '/') + ".class";
+ ClassLoader loader = klass.getClassLoader();
InputStream input = loader.getResourceAsStream(path);
try {
File file = new File(directory, path);
@@ -98,12 +101,12 @@ class ForkClient {
}
public synchronized Object echo(Object message) throws IOException {
- consumeErrors();
+ consumeErrorStream();
output.write(ForkServer.ECHO);
ForkSerializer.serialize(output, message);
output.flush();
- readResponseType();
+ waitForResponse();
try {
return ForkSerializer.deserialize(input, loader).toString();
} catch (ClassNotFoundException e) {
@@ -123,16 +126,24 @@ class ForkClient {
delete(directory);
}
- private byte readResponseType() throws IOException {
+ /**
+ *
+ * @return
+ * @throws IOException
+ */
+ private byte waitForResponse() throws IOException {
+ List<ForkResource> resources = new ArrayList<ForkResource>();
+ resources.add(new ClassLoaderResource(loader));
while (true) {
- consumeErrors();
+ consumeErrorStream();
int type = input.read();
if (type == -1) {
- throw new IOException("Unexpected end of stream encountered");
- } else if (type == ForkServer.FIND_RESOURCE) {
- sendResource(input.readUTF());
- } else if (type == ForkServer.FIND_RESOURCES) {
- sendResources(input.readUTF());
+ throw new IOException(
+ "Lost connection to a forked server process");
+ } else if (type == ForkServer.RESOURCE) {
+ ForkResource resource =
+ resources.get(input.readUnsignedByte());
+ resource.process(input, output);
} else {
return (byte) type;
}
@@ -140,73 +151,15 @@ class ForkClient {
}
/**
- * Sends the named resource to the forked server process over the
- * stdin/out communication channel. The resource stream is preceded
- * with a boolean <code>true</code> value if the resource was found,
- * otherwise just a boolean <code>false</code> value is written.
- *
- * @param name resource name
- * @throws IOException if the resource could not be sent
- */
- private void sendResource(String name) throws IOException {
- InputStream stream = loader.getResourceAsStream(name);
- if (stream != null) {
- output.writeBoolean(true);
- writeAndCloseStream(stream);
- } else {
- output.writeBoolean(false);
- }
- output.flush();
- }
-
- /**
- * Sends all the named resources to the forked server process over the
- * stdin/out communication channel. Each resource stream is preceded
- * with a boolean <code>true</code> value, and a single boolean
- * <code>false</code> value is written when no longer resources
- * are available.
- *
- * @param name resource name
- * @throws IOException if the resources could not be sent
- */
- private void sendResources(String name) throws IOException {
- Enumeration<URL> resources = loader.getResources(name);
- while (resources.hasMoreElements()) {
- output.writeBoolean(true);
- writeAndCloseStream(resources.nextElement().openStream());
- }
- output.writeBoolean(false);
- output.flush();
- }
-
- /**
- * Sends the given byte stream to the forked server process over the
- * stdin/out communication channel. The stream is sent in chunks of
- * less than 64kB, each preceded by a short value that indicates the
- * length of the following chunk. A zero short value is sent at the
- * end to signify the end of the stream.
- * <p>
- * The stream is guaranteed to be closed by this method, regardless of
- * the way it returns.
+ * Consumes all pending bytes from the standard error stream of the
+ * forked server process, and prints them out to the standard error
+ * stream of this process. This method should be called always before
+ * expecting some output from the server, to prevent the server from
+ * blocking due to a filled up pipe buffer of the error stream.
*
- * @param stream the stream to be sent
- * @throws IOException if the stream could not be sent
+ * @throws IOException if the error stream could not be read
*/
- private void writeAndCloseStream(InputStream stream) throws IOException {
- try {
- byte[] buffer = new byte[0xffff];
- int n;
- while ((n = stream.read(buffer)) != -1) {
- output.writeShort(n);
- output.write(buffer, 0, n);
- }
- output.writeShort(0);
- } finally {
- stream.close();
- }
- }
-
- private void consumeErrors() throws IOException {
+ private void consumeErrorStream() throws IOException {
int n;
while ((n = error.available()) > 0) {
byte[] b = new byte[n];
Added: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkResource.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkResource.java?rev=1059879&view=auto
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkResource.java (added)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkResource.java Mon Jan 17 11:22:56 2011
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.fork;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+import java.io.IOException;
+
+interface ForkResource {
+
+ void process(DataInputStream input, DataOutputStream output)
+ throws IOException;
+
+}
Modified: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java?rev=1059879&r1=1059878&r2=1059879&view=diff
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java (original)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java Mon Jan 17 11:22:56 2011
@@ -39,9 +39,7 @@ class ForkServer extends ClassLoader {
public static final byte ECHO = 1;
- public static final byte FIND_RESOURCE = 2;
-
- public static final byte FIND_RESOURCES = 3;
+ public static final byte RESOURCE = 2;
/**
* Starts a forked server process.
@@ -108,7 +106,9 @@ class ForkServer extends ClassLoader {
protected synchronized URL findResource(String name) {
try {
// Send a request to load the resource data
- output.write(FIND_RESOURCE);
+ output.write(RESOURCE);
+ output.write(0);
+ output.write(1);
output.writeUTF(name);
output.flush();
@@ -127,7 +127,9 @@ class ForkServer extends ClassLoader {
protected synchronized Enumeration<URL> findResources(String name)
throws IOException {
// Send a request to load the resources
- output.write(FIND_RESOURCE);
+ output.write(RESOURCE);
+ output.write(0);
+ output.write(2);
output.writeUTF(name);
output.flush();
@@ -144,7 +146,9 @@ class ForkServer extends ClassLoader {
throws ClassNotFoundException {
try {
// Send a request to load the class data
- output.write(FIND_RESOURCE);
+ output.write(RESOURCE);
+ output.write(0);
+ output.write(1);
output.writeUTF(name.replace('.', '/') + ".class");
output.flush();