@@ -0,0 +1,2830 @@
+#%0A# Licensed to the Apache Software Foundation (ASF) under one or more%0A# contributor license agreements.  See the NOTICE file distributed with%0A# this work for additional information regarding copyright ownership.%0A# The ASF licenses this file to You under the Apache License, Version 2.0%0A# (the %22License%22); you may not use this file except in compliance with%0A# the License.  You may obtain a copy of the License at%0A#%0A#    http://www.apache.org/licenses/LICENSE-2.0%0A#%0A# Unless required by applicable law or agreed to in writing, software%0A# distributed under the License is distributed on an %22AS IS%22 BASIS,%0A# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A# See the License for the specific language governing permissions and%0A# limitations under the License.%0A#%0A# pytype: skip-file%0A%0Aimport argparse%0Aimport logging%0Aimport signal%0Aimport sys%0A%0Aimport grpc%0A%0Afrom apache_beam.pipeline import PipelineOptions%0Afrom apache_beam.portability.api import beam_artifact_api_pb2_grpc%0Afrom apache_beam.portability.api import beam_expansion_api_pb2_grpc%0Afrom apache_beam.runners.portability import artifact_service%0Afrom apache_beam.runners.portability import expansion_service%0Afrom apache_beam.transforms import fully_qualified_named_transform%0Afrom apache_beam.utils import thread_pool_executor%0A%0A_LOGGER = logging.getLogger(__name__)%0A%0A%0Adef main(argv):%0A  parser = argparse.ArgumentParser()%0A  parser.add_argument(%0A      '-p', '--port', type=int, help='port on which to serve the job api')%0A  parser.add_argument('--fully_qualified_name_glob', default=None)%0A  known_args, pipeline_args = parser.parse_known_args(argv)%0A  pipeline_options = PipelineOptions(%0A      pipeline_args + %5B%22--experiments=beam_fn_api%22, %22--sdk_location=container%22%5D)%0A%0A  with fully_qualified_named_transform.FullyQualifiedNamedTransform.with_filter(%0A      known_args.fully_qualified_name_glob):%0A%0A    server = grpc.server(thread_pool_executor.shared_unbounded_instance())%0A    beam_expansion_api_pb2_grpc.add_ExpansionServiceServicer_to_server(%0A        expansion_service.ExpansionServiceServicer(pipeline_options), server)%0A    beam_artifact_api_pb2_grpc.add_ArtifactRetrievalServiceServicer_to_server(%0A        artifact_service.ArtifactRetrievalService(%0A            artifact_service.BeamFilesystemHandler(None).file_reader),%0A        server)%0A    server.add_insecure_port('localhost:%7B%7D'.format(known_args.port))%0A    server.start()%0A    _LOGGER.info('Listening for expansion requests at %25d', known_args.port)%0A%0A    def cleanup(unused_signum, unused_frame):%0A      _LOGGER.info('Shutting down expansion service.')%0A      server.stop(None)%0A%0A    signal.signal(signal.SIGTERM, cleanup)%0A    signal.signal(signal.SIGINT, cleanup)%0A    # blocking main thread forever.%0A    signal.pause()%0A%0A%0Aif __name__ == '__main__':%0A  logging.getLogger().setLevel(logging.INFO)%0A  main(sys.argv)%0A

Add Python expansion service entry point.