(self, seqhdr, body)
| 110 | return True |
| 111 | |
| 112 | def process_message(self, seqhdr, body): |
| 113 | typeid = nodeid_from_binary(body) |
| 114 | requesthdr = struct_from_binary(ua.RequestHeader, body) |
| 115 | try: |
| 116 | return self._process_message(typeid, requesthdr, seqhdr, body) |
| 117 | except utils.ServiceError as e: |
| 118 | status = ua.StatusCode(e.code) |
| 119 | response = ua.ServiceFault() |
| 120 | response.ResponseHeader.ServiceResult = status |
| 121 | self.logger.info("sending service fault response: %s (%s)", status.doc, status.name) |
| 122 | self.send_response(requesthdr.RequestHandle, seqhdr, response) |
| 123 | return True |
| 124 | |
| 125 | def _process_message(self, typeid, requesthdr, seqhdr, body): |
| 126 | if typeid == ua.NodeId(ua.ObjectIds.CreateSessionRequest_Encoding_DefaultBinary): |
no test coverage detected