Tika gRPC

Usage: <main class> [options]
  Options:
    --cert-chain
      Certificate chain file. Example: server1.pem See: https://github.com/grpc/grpc-java/tree/b3ffb5078df361d7460786e134db7b5c00939246/examples/example-tls
    --client-auth-required
      Is Mutual TLS required? Implies --secure.
      Default: false
    -c, --config
      The tika config file
    -h, -H, --help
      Display help menu
      Default: false
    --plugin-roots
      Comma-separated list of plugin root directories (overrides config file)
    -l, --plugins
      The tika pipes plugins config file
    -p, --port
      The grpc server port
    --private-key
      Private key store. Example: server1.key See: https://github.com/grpc/grpc-java/tree/b3ffb5078df361d7460786e134db7b5c00939246/examples/example-tls
    --private-key-password
      Private key password, if needed
    -s, --secure
      Enable credentials required to access this grpc server
      Default: false
    --trust-cert-collection
      The trust certificate collection (root certs). Required, and must be a
      readable file, when --client-auth-required is set. Example: ca.pem See: https://github.com/grpc/grpc-java/tree/b3ffb5078df361d7460786e134db7b5c00939246/examples/example-tls

message FetchAndParseRequest {
  string fetcher_id = 1;                    // a fetcher from the config or the fetcher store
  string fetch_key = 2;                     // e.g. a path relative to basePath, a URL, an object key
  string additional_fetch_config_json = 3;  // per-request fetcher overrides (allowPerRequestConfig)
  string emitter_id = 4;                    // see the note below
  string parse_context_json = 5;            // per-request ParseContext (allowPerRequestConfig)
}

message FetchAndParseReply {
  string fetch_key = 1;                     // echoes the request
  map<string, string> fields = 2;           // metadata, including tk:content
  string status = 3;                        // a PipesResult.RESULT_STATUS name
  string error_message = 4;                 // set for FETCH_EXCEPTION
}

{
  "plugin-roots": "/opt/bookshelf/plugins",
  "fetchers": {
    "bookshelf-inbox": {
      "file-system-fetcher": { "basePath": "/srv/bookshelf/inbox" }
    }
  },
  "pipes": {
    "numClients": 2,
    "forkedJvmArgs": ["-Xmx1g"],
    "emitStrategy": { "type": "PASSBACK_ALL" }
  },
  "parse-context": {
    "basic-content-handler-factory": { "type": "MARKDOWN", "writeLimit": 1000000 }
  },
  "grpc": {
    "allowPerRequestConfig": false,
    "allowComponentManagement": false
  }
}

CLIENT_UNAVAILABLE_WITHIN_MS fields=0 contentChars=0
CLIENT_UNAVAILABLE_WITHIN_MS fields=0 contentChars=0
CLIENT_UNAVAILABLE_WITHIN_MS fields=0 contentChars=0
PARSE_SUCCESS fields=16 contentChars=9073247

PERMISSION_DENIED: Per-request configuration is disabled. Set 'allowPerRequestConfig' to true in the 'grpc' section of
your tika-config to allow additional_fetch_config_json / parse_context_json. Understand the security implications first.
PERMISSION_DENIED: Runtime component management is disabled. Set 'allowComponentManagement' to true in the 'grpc'
section of your tika-config to allow SaveFetcher/DeleteFetcher/SavePipesIterator/DeletePipesIterator. Understand the
security implications first.

package demo;

import java.util.concurrent.TimeUnit;

import io.grpc.Grpc;
import io.grpc.InsecureChannelCredentials;
import io.grpc.ManagedChannel;
import io.grpc.StatusRuntimeException;

import org.apache.tika.pipes.grpc.proto.FetchAndParseReply;
import org.apache.tika.pipes.grpc.proto.FetchAndParseRequest;
import org.apache.tika.pipes.grpc.proto.TikaGrpc;

public class PerRequestContextClient {

    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = Grpc.newChannelBuilder(args[0], InsecureChannelCredentials.create())
                .maxInboundMessageSize(32 * 1024 * 1024)
                .build();
        TikaGrpc.TikaBlockingStub tika = TikaGrpc.newBlockingStub(channel);
        String[] contexts = {
            "{\"parse-mode\": \"CONCATENATE\", \"basic-content-handler-factory\": {\"type\": \"TEXT\"}}",
            "{\"parse-mode\": \"NO_PARSE\"}",
            "{\"exception-reporting\": {\"level\": \"FULL\"}}",
            "{\"basic-content-handler-factroy\": {\"type\": \"TEXT\"}}"
        };
        try {
            for (String json : contexts) {
                try {
                    FetchAndParseReply reply = tika.withDeadlineAfter(2, TimeUnit.MINUTES)
                            .fetchAndParse(FetchAndParseRequest.newBuilder()
                                    .setFetcherId("bookshelf-inbox")
                                    .setFetchKey("catalogue-email.eml")
                                    .setParseContextJson(json)
                                    .build());
                    System.out.println(reply.getStatus() + " Content-Type=" + reply.getFieldsMap().get("Content-Type")
                            + " handler=" + reply.getFieldsMap().get("tk:content-handler-type")
                            + " fields=" + reply.getFieldsCount());
                } catch (StatusRuntimeException e) {
                    System.out.println(e.getStatus().getCode() + ": " + e.getStatus().getDescription());
                }
            }
        } finally {
            channel.shutdown().awaitTermination(10, TimeUnit.SECONDS);
        }
    }
}

PARSE_SUCCESS Content-Type=message/rfc822 handler=TEXT fields=23
PARSE_SUCCESS Content-Type=message/rfc822 handler=null fields=6
INVALID_ARGUMENT: Invalid parse_context_json: Component 'exception-reporting' (context key org.apache.tika.config.ExceptionReporting) may not be supplied via a request parseContext. Components of this type must be configured server-side in the Tika config, not at request time.
INVALID_ARGUMENT: Invalid parse_context_json: Unrecognized parse-context entry 'basic-content-handler-factroy'. Check for a typo; it does not match any registered component.

  Content-Type = application/pdf
  dc:title = Spring catalogue
  dc:subject = Spring catalogue and manuscript
  message:from = Marta Ruiz <marta@example-publisher.test>
  tk:content = Spring catalogue  ||Bookshelf spring catalogue Forty titles from independent publishers.|
  tk:content-type-parser-override = message/rfc822
  tk:embedded-depth = 1
  tk:embedded-resource-path = /catalogue.pdf
  tk:resource-name = catalogue.pdf
  tk:source-path = catalogue-email.eml

package demo;

import java.util.concurrent.TimeUnit;

import io.grpc.Grpc;
import io.grpc.InsecureChannelCredentials;
import io.grpc.ManagedChannel;

import org.apache.tika.pipes.grpc.proto.FetchAndParseReply;
import org.apache.tika.pipes.grpc.proto.FetchAndParseRequest;
import org.apache.tika.pipes.grpc.proto.GetFetcherConfigJsonSchemaRequest;
import org.apache.tika.pipes.grpc.proto.GetFetcherReply;
import org.apache.tika.pipes.grpc.proto.GetFetcherRequest;
import org.apache.tika.pipes.grpc.proto.ListFetchersRequest;
import org.apache.tika.pipes.grpc.proto.SaveFetcherRequest;
import org.apache.tika.pipes.grpc.proto.TikaGrpc;

public class FetcherAdminClient {

    public static void main(String[] args) throws InterruptedException {
        String target = args[0];
        String uploadsDir = args[1]; // a directory the server can read
        ManagedChannel channel = Grpc.newChannelBuilder(target, InsecureChannelCredentials.create()).build();
        try {
            TikaGrpc.TikaBlockingStub tika = TikaGrpc.newBlockingStub(channel)
                    .withDeadlineAfter(2, TimeUnit.MINUTES);

            String schema = tika.getFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest.newBuilder()
                    .setFetcherType("file-system-fetcher").build()).getFetcherConfigJsonSchema();
            System.out.println("schema: " + schema.replaceAll("\\s+", " "));

            tika.saveFetcher(SaveFetcherRequest.newBuilder()
                    .setFetcherId("bookshelf-uploads")
                    .setFetcherType("file-system-fetcher")
                    .setFetcherConfigJson("{\"basePath\": \"" + uploadsDir + "\"}")
                    .build());

            GetFetcherReply saved = tika.getFetcher(GetFetcherRequest.newBuilder()
                    .setFetcherId("bookshelf-uploads").build());
            System.out.println("saved: " + saved.getFetcherId() + " " + saved.getFetcherType()
                    + " " + saved.getParamsMap().keySet());

            tika.listFetchers(ListFetchersRequest.newBuilder().setNumFetchersPerPage(50).build())
                    .getGetFetcherRepliesList()
                    .forEach(f -> System.out.println("listed: " + f.getFetcherId()));

            FetchAndParseReply reply = tika.fetchAndParse(FetchAndParseRequest.newBuilder()
                    .setFetcherId("bookshelf-uploads").setFetchKey("catalogue.pdf").build());
            System.out.println("parse: " + reply.getStatus());
        } finally {
            channel.shutdown().awaitTermination(10, TimeUnit.SECONDS);
        }
    }
}

schema: { "type" : "object", "id" : "urn:jsonschema:org:apache:tika:pipes:fetcher:fs:FileSystemFetcherConfig", "properties" : { "basePath" : { "type" : "string" }, "extractFileSystemMetadata" : { "type" : "boolean" }, "allowAbsolutePaths" : { "type" : "boolean" } } }
saved: bookshelf-uploads file-system-fetcher [basePath]
listed: bookshelf-inbox
listed: bookshelf-uploads
parse: PARSE_SUCCESS

java -cp "$(cat cp.txt)" org.apache.tika.pipes.grpc.TikaGrpcServer -c tika-config.json -p 9443 \
  --client-auth-required \
  --cert-chain certs/server.pem --private-key certs/server.key \
  --trust-cert-collection certs/ca.pem

INFO  org.apache.tika.pipes.grpc.TikaGrpcServer --client-auth-required implies --secure; enabling TLS.
INFO  org.apache.tika.pipes.grpc.TikaGrpcServer Server started, listening on 9443

package demo;

import java.io.File;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

import io.grpc.ChannelCredentials;
import io.grpc.Grpc;
import io.grpc.ManagedChannel;
import io.grpc.StatusRuntimeException;
import io.grpc.TlsChannelCredentials;

import org.apache.tika.pipes.grpc.proto.FetchAndParseReply;
import org.apache.tika.pipes.grpc.proto.FetchAndParseRequest;
import org.apache.tika.pipes.grpc.proto.TikaGrpc;

public class BookshelfTlsClient {

    // args: target ca.pem [client.pem client.key]
    public static void main(String[] args) throws IOException, InterruptedException {
        TlsChannelCredentials.Builder tls = TlsChannelCredentials.newBuilder()
                .trustManager(new File(args[1]));
        if (args.length > 3) {
            tls.keyManager(new File(args[2]), new File(args[3]));
        }
        ChannelCredentials credentials = tls.build();
        ManagedChannel channel = Grpc.newChannelBuilder(args[0], credentials).build();
        try {
            FetchAndParseReply reply = TikaGrpc.newBlockingStub(channel)
                    .withDeadlineAfter(1, TimeUnit.MINUTES)
                    .fetchAndParse(FetchAndParseRequest.newBuilder()
                            .setFetcherId("bookshelf-inbox").setFetchKey("catalogue.pdf").build());
            System.out.println(reply.getStatus() + " " + reply.getFieldsMap().get("dc:title"));
        } catch (StatusRuntimeException e) {
            System.out.println("failed: " + e.getStatus().getCode() + " " + e.getStatus().getDescription());
        } finally {
            channel.shutdownNow().awaitTermination(10, TimeUnit.SECONDS);
        }
    }
}

# with client.pem and client.key
PARSE_SUCCESS Spring catalogue
# with the CA only, no client certificate
failed: UNAVAILABLE ssl exception
# plaintext client (InsecureChannelCredentials)
io.grpc.StatusRuntimeException: UNAVAILABLE: Network closed for unknown reason