]>
Commit | Line | Data |
---|---|---|
f67539c2 TL |
1 | /** |
2 | * Licensed to the Apache Software Foundation (ASF) under one | |
3 | * or more contributor license agreements. See the NOTICE file | |
4 | * distributed with this work for additional information | |
5 | * regarding copyright ownership. The ASF licenses this file | |
6 | * to you under the Apache License, Version 2.0 (the | |
7 | * "License"); you may not use this file except in compliance | |
8 | * with the License. You may obtain a copy of the License at | |
9 | * | |
10 | * http://www.apache.org/licenses/LICENSE-2.0 | |
11 | * | |
12 | * Unless required by applicable law or agreed to in writing, | |
13 | * software distributed under the License is distributed on an | |
14 | * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
15 | * KIND, either express or implied. See the License for the | |
16 | * specific language governing permissions and limitations | |
17 | * under the License. | |
18 | */ | |
19 | ||
20 | package org.apache.thrift.protocol; | |
21 | ||
22 | import haxe.ds.StringMap; | |
23 | import org.apache.thrift.TApplicationException; | |
24 | import org.apache.thrift.TProcessor; | |
25 | ||
26 | import org.apache.thrift.transport.TTransport; | |
27 | ||
28 | ||
29 | /** | |
30 | * TMultiplexedProcessor is a TProcessor allowing a single TServer to provide multiple services. | |
31 | * To do so, you instantiate the processor and then register additional processors with it, | |
32 | * as shown in the following example: | |
33 | * | |
34 | * TMultiplexedProcessor processor = new TMultiplexedProcessor(); | |
35 | * | |
36 | * processor.registerProcessor( | |
37 | * "Calculator", | |
38 | * new Calculator.Processor(new CalculatorHandler())); | |
39 | * | |
40 | * processor.registerProcessor( | |
41 | * "WeatherReport", | |
42 | * new WeatherReport.Processor(new WeatherReportHandler())); | |
43 | * | |
44 | * TServerTransport t = new TServerSocket(9090); | |
45 | * TSimpleServer server = new TSimpleServer(processor, t); | |
46 | * | |
47 | * server.serve(); | |
48 | */ | |
49 | class TMultiplexedProcessor implements TProcessor | |
50 | { | |
51 | private var serviceProcessorMap : StringMap<TProcessor> = new StringMap<TProcessor>(); | |
52 | private var defaultProcessor : TProcessor = null; | |
53 | ||
54 | public function new() { | |
55 | } | |
56 | ||
57 | /** | |
58 | * 'Register' a service with this TMultiplexedProcessor. This allows us to broker | |
59 | * requests to individual services by using the service name to select them at request time. | |
60 | * | |
61 | * Args: | |
62 | * - serviceName Name of a service, has to be identical to the name | |
63 | * declared in the Thrift IDL, e.g. "WeatherReport". | |
64 | * - processor Implementation of a service, usually referred to as "handlers", | |
65 | * e.g. WeatherReportHandler implementing WeatherReport.Iface. | |
66 | */ | |
67 | public function RegisterProcessor(serviceName : String, processor : TProcessor, asDefault : Bool = false) : Void { | |
68 | serviceProcessorMap.set(serviceName, processor); | |
69 | if ( asDefault) { | |
70 | if( defaultProcessor != null) { | |
71 | throw new TApplicationException( TApplicationException.UNKNOWN, "Can't have multiple default processors"); | |
72 | } else { | |
73 | defaultProcessor = processor; | |
74 | } | |
75 | } | |
76 | } | |
77 | ||
78 | ||
79 | private function Fail( oprot : TProtocol, message : TMessage, extype : Int, etxt : String) : Void { | |
80 | var appex = new TApplicationException( extype, etxt); | |
81 | ||
82 | var newMessage = new TMessage(message.name, TMessageType.EXCEPTION, message.seqid); | |
83 | ||
84 | oprot.writeMessageBegin(newMessage); | |
85 | appex.write( oprot); | |
86 | oprot.writeMessageEnd(); | |
87 | oprot.getTransport().flush(); | |
88 | } | |
89 | ||
90 | ||
91 | /** | |
92 | * This implementation of process performs the following steps: | |
93 | * | |
94 | * - Read the beginning of the message. | |
95 | * - Extract the service name from the message. | |
96 | * - Using the service name to locate the appropriate processor. | |
97 | * - Dispatch to the processor, with a decorated instance of TProtocol | |
98 | * that allows readMessageBegin() to return the original TMessage. | |
99 | * | |
100 | * Throws an exception if | |
101 | * - the message type is not CALL or ONEWAY, | |
102 | * - the service name was not found in the message, or | |
103 | * - the service name has not been RegisterProcessor()ed. | |
104 | */ | |
105 | public function process( iprot : TProtocol, oprot : TProtocol) : Bool { | |
106 | /* Use the actual underlying protocol (e.g. TBinaryProtocol) to read the | |
107 | message header. This pulls the message "off the wire", which we'll | |
108 | deal with at the end of this method. */ | |
109 | ||
110 | var message : TMessage = iprot.readMessageBegin(); | |
111 | var methodName : String = ""; | |
112 | ||
113 | if ((message.type != TMessageType.CALL) && (message.type != TMessageType.ONEWAY)) | |
114 | { | |
115 | Fail(oprot, message, | |
116 | TApplicationException.INVALID_MESSAGE_TYPE, | |
117 | "Message type CALL or ONEWAY expected"); | |
118 | return false; | |
119 | } | |
120 | ||
121 | // Extract the service name | |
122 | var actualProcessor : TProcessor = null; | |
123 | var index = message.name.indexOf(TMultiplexedProtocol.SEPARATOR); | |
124 | if (index < 0) { | |
125 | // fallback to default processor | |
126 | methodName = message.name; | |
127 | actualProcessor = defaultProcessor; | |
128 | if( actualProcessor == null) { | |
129 | Fail(oprot, message, | |
130 | TApplicationException.INVALID_PROTOCOL, | |
131 | "Service name not found in message name: " + message.name + " and no default processor defined. " + | |
132 | "Did you forget to use a TMultiplexProtocol in your client?"); | |
133 | return false; | |
134 | } | |
135 | ||
136 | } else { | |
137 | // service name given | |
138 | var serviceName = message.name.substring(0, index); | |
139 | methodName = message.name.substring( serviceName.length + TMultiplexedProtocol.SEPARATOR.length); | |
140 | actualProcessor = serviceProcessorMap.get( serviceName); | |
141 | if( actualProcessor == null) { | |
142 | Fail(oprot, message, | |
143 | TApplicationException.INTERNAL_ERROR, | |
144 | "Service name not found: " + serviceName + ". " + | |
145 | "Did you forget to call RegisterProcessor()?"); | |
146 | return false; | |
147 | } | |
148 | } | |
149 | ||
150 | // Create a new TMessage, removing the service name | |
151 | // Dispatch processing to the stored processor | |
152 | var newMessage = new TMessage( methodName, message.type, message.seqid); | |
153 | var storedMsg = new StoredMessageProtocol( iprot, newMessage); | |
154 | return actualProcessor.process( storedMsg, oprot); | |
155 | } | |
156 | } | |
157 | ||
158 | ||
159 | /** | |
160 | * Our goal was to work with any protocol. In order to do that, we needed | |
161 | * to allow them to call readMessageBegin() and get a TMessage in exactly | |
162 | * the standard format, without the service name prepended to TMessage.name. | |
163 | */ | |
164 | class StoredMessageProtocol extends TProtocolDecorator | |
165 | { | |
166 | private var messageBegin : TMessage; | |
167 | ||
168 | public function new( protocol : TProtocol, messageBegin : TMessage) { | |
169 | super( protocol); | |
170 | this.messageBegin = messageBegin; | |
171 | } | |
172 | ||
173 | public override function readMessageBegin() : TMessage { | |
174 | return messageBegin; | |
175 | } | |
176 | } | |
177 |