blob: 05563d0794df7add764330860f41c71bc782afd0 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
package org.openslx.thrifthelper;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayDeque;
import java.util.Collections;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.transport.TTransportException;
import org.openslx.thrifthelper.ThriftManager.ErrorCallback;
class ThriftHandler<T extends TServiceClient> implements InvocationHandler
{
private final static Logger LOGGER = Logger.getLogger( ThriftHandler.class );
protected interface WantClientCallback<T>
{
public T getNewClient();
}
private final Deque<T> pool = new ArrayDeque<>();
private final WantClientCallback<? extends T> callback;
private final ErrorCallback errorCallback;
private final Set<String> thriftMethods;
public ThriftHandler( final Class<? extends T> clazz, WantClientCallback<? extends T> cb, ErrorCallback errCb )
{
errorCallback = errCb;
callback = cb;
Set<String> tmpset = new HashSet<String>();
Method[] methods = clazz.getMethods();
for ( int i = 0; i < methods.length; i++ ) {
boolean thrift = false;
Class<?>[] type = methods[i].getExceptionTypes();
for ( int e = 0; e < type.length; e++ ) {
if ( TException.class.isAssignableFrom( type[e] ) )
thrift = true;
}
String name = methods[i].getName();
if ( thrift && !name.startsWith( "send_" ) && !name.startsWith( "recv_" ) ) {
tmpset.add( name );
}
}
thriftMethods = Collections.unmodifiableSet( tmpset );
}
@Override
public Object invoke( Object tproxy, Method method, Object[] args ) throws Throwable
{
// first find the thrift methods
if ( !thriftMethods.contains( method.getName() ) ) {
throw new IllegalAccessException( "Cannot call this method on a proxied thrift client" );
}
T client = getClient();
try {
Throwable cause = null;
for ( int i = 1;; i++ ) {
if ( client != null ) {
try {
return method.invoke( client, args );
} catch ( InvocationTargetException e ) {
cause = e.getCause();
if ( cause != null && ! ( cause instanceof TException ) ) {
throw cause;
}
freeClient( client );
client = null;
if ( cause == null ) {
cause = e;
}
}
}
// Call the error callback. As long as true is returned, keep retrying
if ( !errorCallback.thriftError( i, method.getName(), cause ) ) {
break;
}
if ( client == null ) {
client = getClient();
cause = null;
}
}
// Uh oh
if ( cause != null )
throw cause;
throw new TTransportException( "Could not connect" );
} finally {
returnClient( client );
}
}
private void freeClient( T client )
{
try {
client.getInputProtocol().getTransport().close();
} catch ( Exception e ) {
}
try {
client.getOutputProtocol().getTransport().close();
} catch ( Exception e ) {
}
}
private T getClient()
{
T client;
synchronized ( pool ) {
client = pool.poll();
if ( client != null ) {
return client;
}
}
// Pool is closed, create new client
LOGGER.debug( "Creating new thrift client" );
return callback.getNewClient();
}
private void returnClient( T client )
{
if ( client == null )
return;
synchronized ( pool ) {
pool.push( client );
}
}
}
|
