169 lines · python
1#!/usr/bin/env python32# ex: set filetype=python:3 4"""Generate code for an RPC program's procedures"""5 6from jinja2 import Environment7 8from generators import SourceGenerator, create_jinja2_environment9from xdr_ast import _RpcProgram, _RpcVersion, excluded_apis10 11 12def emit_version_definitions(13 environment: Environment, program: str, version: _RpcVersion14) -> None:15 """Emit procedure numbers for each RPC version's procedures"""16 template = environment.get_template("definition/open.j2")17 print(template.render(program=program.upper()))18 19 template = environment.get_template("definition/procedure.j2")20 for procedure in version.procedures:21 if procedure.name not in excluded_apis:22 print(23 template.render(24 name=procedure.name,25 value=procedure.number,26 )27 )28 29 template = environment.get_template("definition/close.j2")30 print(template.render())31 32 33def emit_version_declarations(34 environment: Environment, program: str, version: _RpcVersion35) -> None:36 """Emit declarations for each RPC version's procedures"""37 arguments = dict.fromkeys([])38 for procedure in version.procedures:39 if procedure.name not in excluded_apis:40 arguments[procedure.argument.type_name] = None41 if len(arguments) > 0:42 print("")43 template = environment.get_template("declaration/argument.j2")44 for argument in arguments:45 print(template.render(program=program, argument=argument))46 47 results = dict.fromkeys([])48 for procedure in version.procedures:49 if procedure.name not in excluded_apis:50 results[procedure.result.type_name] = None51 if len(results) > 0:52 print("")53 template = environment.get_template("declaration/result.j2")54 for result in results:55 print(template.render(program=program, result=result))56 57 58def emit_version_argument_decoders(59 environment: Environment, program: str, version: _RpcVersion60) -> None:61 """Emit server argument decoders for each RPC version's procedures"""62 arguments = dict.fromkeys([])63 for procedure in version.procedures:64 if procedure.name not in excluded_apis:65 arguments[procedure.argument.type_name] = None66 67 template = environment.get_template("decoder/argument.j2")68 for argument in arguments:69 print(template.render(program=program, argument=argument))70 71 72def emit_version_result_decoders(73 environment: Environment, program: str, version: _RpcVersion74) -> None:75 """Emit client result decoders for each RPC version's procedures"""76 results = dict.fromkeys([])77 for procedure in version.procedures:78 if procedure.name not in excluded_apis:79 results[procedure.result.type_name] = None80 81 template = environment.get_template("decoder/result.j2")82 for result in results:83 print(template.render(program=program, result=result))84 85 86def emit_version_argument_encoders(87 environment: Environment, program: str, version: _RpcVersion88) -> None:89 """Emit client argument encoders for each RPC version's procedures"""90 arguments = dict.fromkeys([])91 for procedure in version.procedures:92 if procedure.name not in excluded_apis:93 arguments[procedure.argument.type_name] = None94 95 template = environment.get_template("encoder/argument.j2")96 for argument in arguments:97 print(template.render(program=program, argument=argument))98 99 100def emit_version_result_encoders(101 environment: Environment, program: str, version: _RpcVersion102) -> None:103 """Emit server result encoders for each RPC version's procedures"""104 results = dict.fromkeys([])105 for procedure in version.procedures:106 if procedure.name not in excluded_apis:107 results[procedure.result.type_name] = None108 109 template = environment.get_template("encoder/result.j2")110 for result in results:111 print(template.render(program=program, result=result))112 113 114class XdrProgramGenerator(SourceGenerator):115 """Generate source code for an RPC program's procedures"""116 117 def __init__(self, language: str, peer: str):118 """Initialize an instance of this class"""119 self.environment = create_jinja2_environment(language, "program")120 self.peer = peer121 122 def emit_definition(self, node: _RpcProgram) -> None:123 """Emit procedure numbers for each of an RPC programs's procedures"""124 raw_name = node.name125 program = raw_name.lower().removesuffix("_program").removesuffix("_prog")126 127 for version in node.versions:128 emit_version_definitions(self.environment, program, version)129 130 def emit_declaration(self, node: _RpcProgram) -> None:131 """Emit a declaration pair for each of an RPC programs's procedures"""132 raw_name = node.name133 program = raw_name.lower().removesuffix("_program").removesuffix("_prog")134 135 for version in node.versions:136 emit_version_declarations(self.environment, program, version)137 138 def emit_decoder(self, node: _RpcProgram) -> None:139 """Emit all decoder functions for an RPC program's procedures"""140 raw_name = node.name141 program = raw_name.lower().removesuffix("_program").removesuffix("_prog")142 match self.peer:143 case "server":144 for version in node.versions:145 emit_version_argument_decoders(146 self.environment, program, version,147 )148 case "client":149 for version in node.versions:150 emit_version_result_decoders(151 self.environment, program, version,152 )153 154 def emit_encoder(self, node: _RpcProgram) -> None:155 """Emit all encoder functions for an RPC program's procedures"""156 raw_name = node.name157 program = raw_name.lower().removesuffix("_program").removesuffix("_prog")158 match self.peer:159 case "server":160 for version in node.versions:161 emit_version_result_encoders(162 self.environment, program, version,163 )164 case "client":165 for version in node.versions:166 emit_version_argument_encoders(167 self.environment, program, version,168 )169