Skip to content

Commit ac83fad

Browse files
mondainmondain
committed
Versioned to 1.3.19; Refactored connection creation and management
1 parent 9beb38c commit ac83fad

File tree

26 files changed

+364
-471
lines changed

26 files changed

+364
-471
lines changed

‎client/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
<parent>
44
<groupId>org.red5</groupId>
55
<artifactId>red5-parent</artifactId>
6-
<version>1.3.18</version>
6+
<version>1.3.19</version>
77
</parent>
88
<modelVersion>4.0.0</modelVersion>
99
<artifactId>red5-client</artifactId>

‎client/src/main/java/org/red5/client/Red5Client.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ public final class Red5Client {
1818
/**
1919
* Current server version with revision
2020
*/
21-
public static final String VERSION = "Red5 Client 1.3.18";
21+
public static final String VERSION = "Red5 Client 1.3.19";
2222

2323
/**
2424
* Create a new Red5Client object using the connection local to the current thread A bit of magic that lets you access the red5 scope

‎client/src/main/java/org/red5/client/net/rtmp/RTMPConnManager.java‎ renamed to ‎client/src/main/java/org/red5/client/net/rtmp/RTMPClientConnManager.java‎

Lines changed: 35 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,13 @@
99

1010
import java.util.ArrayList;
1111
import java.util.Collection;
12+
import java.util.List;
1213
import java.util.concurrent.ConcurrentHashMap;
1314
import java.util.concurrent.ConcurrentMap;
1415
import java.util.concurrent.atomic.AtomicInteger;
1516

1617
import org.red5.client.net.rtmpt.RTMPTClientConnection;
18+
import org.red5.server.BaseConnection;
1719
import org.red5.server.api.Red5;
1820
import org.red5.server.net.IConnectionManager;
1921
import org.red5.server.net.rtmp.RTMPConnection;
@@ -27,9 +29,9 @@
2729
*
2830
* @author The Red5 Project
2931
*/
30-
public class RTMPConnManager implements IConnectionManager<RTMPConnection> {
32+
public class RTMPClientConnManager implements IConnectionManager<BaseConnection> {
3133

32-
private static final Logger log = LoggerFactory.getLogger(RTMPConnManager.class);
34+
private static final Logger log = LoggerFactory.getLogger(RTMPClientConnManager.class);
3335

3436
private static int maxHandshakeTimeout = 7000;
3537

@@ -42,21 +44,24 @@ public class RTMPConnManager implements IConnectionManager<RTMPConnection> {
4244
// whether or not to use the ThreadPoolTaskExecutor for incoming messages
4345
protected static boolean enableTaskExecutor;
4446

45-
protected static IConnectionManager<RTMPConnection> instance = new RTMPConnManager();
47+
protected static IConnectionManager<BaseConnection> instance;
4648

47-
protected ConcurrentMap<String, RTMPConnection> connMap = new ConcurrentHashMap<>();
49+
protected ConcurrentMap<String, BaseConnection> connMap = new ConcurrentHashMap<>();
4850

4951
protected AtomicInteger conns = new AtomicInteger();
5052

51-
public static IConnectionManager<RTMPConnection> getInstance() {
53+
public static IConnectionManager<BaseConnection> getInstance() {
54+
if (instance == null) {
55+
instance = new RTMPClientConnManager();
56+
}
5257
return instance;
5358
}
5459

5560
/** {@inheritDoc} */
5661
@Override
57-
public RTMPConnection createConnection(Class<?> connCls) {
58-
RTMPConnection conn = null;
59-
if (RTMPConnection.class.isAssignableFrom(connCls)) {
62+
public BaseConnection createConnection(Class<?> connCls) {
63+
BaseConnection conn = null;
64+
if (BaseConnection.class.isAssignableFrom(connCls)) {
6065
try {
6166
// create connection
6267
conn = createConnectionInstance(connCls);
@@ -73,9 +78,9 @@ public RTMPConnection createConnection(Class<?> connCls) {
7378

7479
/** {@inheritDoc} */
7580
@Override
76-
public RTMPConnection createConnection(Class<?> connCls, String sessionId) {
77-
RTMPConnection conn = null;
78-
if (RTMPConnection.class.isAssignableFrom(connCls)) {
81+
public BaseConnection createConnection(Class<?> connCls, String sessionId) {
82+
BaseConnection conn = null;
83+
if (BaseConnection.class.isAssignableFrom(connCls)) {
7984
try {
8085
// create connection
8186
conn = createConnectionInstance(connCls);
@@ -93,47 +98,14 @@ public RTMPConnection createConnection(Class<?> connCls, String sessionId) {
9398
return conn;
9499
}
95100

96-
/**
97-
* Adds a connection.
98-
*
99-
* @param conn connection
100-
*/
101-
@Override
102-
public void setConnection(RTMPConnection conn) {
103-
log.trace("Adding connection: {}", conn);
104-
int id = conn.getId();
105-
if (id == -1) {
106-
log.debug("Connection has unsupported id, using session id hash");
107-
id = conn.getSessionId().hashCode();
108-
}
109-
log.debug("Connection id: {} session id hash: {}", conn.getId(), conn.getSessionId().hashCode());
110-
}
111-
112-
/**
113-
* Returns a connection for a given client id.
114-
*
115-
* @param clientId client id
116-
* @return connection if found and null otherwise
117-
*/
118-
@Override
119-
public RTMPConnection getConnection(int clientId) {
120-
log.trace("Getting connection by client id: {}", clientId);
121-
for (RTMPConnection conn : connMap.values()) {
122-
if (conn.getId() == clientId) {
123-
return connMap.get(conn.getSessionId());
124-
}
125-
}
126-
return null;
127-
}
128-
129101
/**
130102
* Returns a connection for a given session id.
131103
*
132104
* @param sessionId session id
133105
* @return connection if found and null otherwise
134106
*/
135107
@Override
136-
public RTMPConnection getConnectionBySessionId(String sessionId) {
108+
public BaseConnection getConnectionBySessionId(String sessionId) {
137109
log.debug("Getting connection by session id: {}", sessionId);
138110
if (connMap.containsKey(sessionId)) {
139111
return connMap.get(sessionId);
@@ -148,28 +120,20 @@ public RTMPConnection getConnectionBySessionId(String sessionId) {
148120

149121
/** {@inheritDoc} */
150122
@Override
151-
public RTMPConnection removeConnection(int clientId) {
152-
log.trace("Removing connection with id: {}", clientId);
153-
// remove from map
154-
for (RTMPConnection conn : connMap.values()) {
155-
if (conn.getId() == clientId) {
156-
// remove the conn
157-
return removeConnection(conn.getSessionId());
158-
}
159-
}
160-
log.warn("Connection was not removed by id: {}", clientId);
161-
return null;
123+
public BaseConnection removeConnection(BaseConnection conn) {
124+
log.trace("Removing connection: {}", conn);
125+
return removeConnection(conn.getSessionId());
162126
}
163127

164128
/** {@inheritDoc} */
165129
@Override
166-
public RTMPConnection removeConnection(String sessionId) {
130+
public BaseConnection removeConnection(String sessionId) {
167131
log.debug("Removing connection with session id: {}", sessionId);
168132
if (log.isTraceEnabled()) {
169133
log.trace("Connections ({}) at pre-remove: {}", connMap.size(), connMap.values());
170134
}
171135
// remove from map
172-
RTMPConnection conn = connMap.remove(sessionId);
136+
BaseConnection conn = connMap.remove(sessionId);
173137
if (conn != null) {
174138
log.trace("Connections: {}", conns.decrementAndGet());
175139
Red5.setConnectionLocal(null);
@@ -179,19 +143,20 @@ public RTMPConnection removeConnection(String sessionId) {
179143

180144
/** {@inheritDoc} */
181145
@Override
182-
public Collection<RTMPConnection> getAllConnections() {
183-
ArrayList<RTMPConnection> list = new ArrayList<RTMPConnection>(connMap.size());
146+
public Collection<BaseConnection> getAllConnections() {
147+
ArrayList<BaseConnection> list = new ArrayList<>(connMap.size());
184148
list.addAll(connMap.values());
185149
return list;
186150
}
187151

188152
/** {@inheritDoc} */
189153
@Override
190-
public Collection<RTMPConnection> removeConnections() {
191-
ArrayList<RTMPConnection> list = new ArrayList<RTMPConnection>(connMap.size());
192-
list.addAll(connMap.values());
193-
connMap.clear();
194-
conns.set(0);
154+
public Collection<BaseConnection> removeConnections() {
155+
final List<BaseConnection> list = new ArrayList<>(connMap.size());
156+
connMap.values().forEach(conn -> {
157+
removeConnection(conn.getSessionId());
158+
list.add(conn);
159+
});
195160
return list;
196161
}
197162

@@ -228,23 +193,23 @@ public RTMPConnection createConnectionInstance(Class<?> cls) throws Exception {
228193
}
229194

230195
public static void setMaxHandshakeTimeout(int maxHandshakeTimeout) {
231-
RTMPConnManager.maxHandshakeTimeout = maxHandshakeTimeout;
196+
RTMPClientConnManager.maxHandshakeTimeout = maxHandshakeTimeout;
232197
}
233198

234199
public static void setMaxInactivity(int maxInactivity) {
235-
RTMPConnManager.maxInactivity = maxInactivity;
200+
RTMPClientConnManager.maxInactivity = maxInactivity;
236201
}
237202

238203
public static void setPingInterval(int pingInterval) {
239-
RTMPConnManager.pingInterval = pingInterval;
204+
RTMPClientConnManager.pingInterval = pingInterval;
240205
}
241206

242207
public static void setExecutorQueueCapacity(int executorQueueCapacity) {
243-
RTMPConnManager.executorQueueCapacity = executorQueueCapacity;
208+
RTMPClientConnManager.executorQueueCapacity = executorQueueCapacity;
244209
}
245210

246211
public static void setEnableTaskExecutor(boolean enableTaskExecutor) {
247-
RTMPConnManager.enableTaskExecutor = enableTaskExecutor;
212+
RTMPClientConnManager.enableTaskExecutor = enableTaskExecutor;
248213
}
249214

250215
}

‎client/src/main/java/org/red5/client/net/rtmp/RTMPMinaIoHandler.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.apache.mina.core.session.IoSession;
1616
import org.red5.client.net.rtmpe.RTMPEClient;
1717
import org.red5.client.net.rtmpe.RTMPEIoFilter;
18+
import org.red5.server.BaseConnection;
1819
import org.red5.server.api.Red5;
1920
import org.red5.server.net.IConnectionManager;
2021
import org.red5.server.net.rtmp.RTMPConnection;
@@ -75,8 +76,7 @@ public void sessionCreated(IoSession session) throws Exception {
7576
// set a reference to the connection on the client
7677
handler.setConnection((RTMPConnection) conn);
7778
// set a connection manager for any required handling and to prevent memory leaking
78-
RTMPConnManager connManager = (RTMPConnManager) RTMPConnManager.getInstance();
79-
session.setAttribute(RTMPConnection.RTMP_CONN_MANAGER, new WeakReference<IConnectionManager<RTMPConnection>>(connManager));
79+
session.setAttribute(RTMPConnection.RTMP_CONN_MANAGER, new WeakReference<IConnectionManager<BaseConnection>>(RTMPClientConnManager.getInstance()));
8080
}
8181

8282
/** {@inheritDoc} */
@@ -175,12 +175,12 @@ public void setEnableSwfVerification(boolean enableSwfVerification) {
175175
}
176176

177177
protected RTMPMinaConnection createRTMPMinaConnection() {
178-
return (RTMPMinaConnection) RTMPConnManager.getInstance().createConnection(RTMPMinaConnection.class);
178+
return (RTMPMinaConnection) RTMPClientConnManager.getInstance().createConnection(RTMPMinaConnection.class);
179179
}
180180

181181
@SuppressWarnings("unchecked")
182-
private RTMPConnManager getConnectionManager(IoSession session) {
183-
return (RTMPConnManager) ((WeakReference<IConnectionManager<RTMPConnection>>) session.getAttribute(RTMPConnection.RTMP_CONN_MANAGER)).get();
182+
private IConnectionManager<RTMPConnection> getConnectionManager(IoSession session) {
183+
return (IConnectionManager<RTMPConnection>) ((WeakReference<IConnectionManager<RTMPConnection>>) session.getAttribute(RTMPConnection.RTMP_CONN_MANAGER)).get();
184184
}
185185

186186
}

‎client/src/main/java/org/red5/client/net/rtmp/codec/RTMPMinaCodecFactory.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
1313
import org.apache.mina.filter.codec.ProtocolEncoder;
1414
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
15-
import org.red5.client.net.rtmp.RTMPConnManager;
15+
import org.red5.client.net.rtmp.RTMPClientConnManager;
1616
import org.red5.server.api.Red5;
1717
import org.red5.server.net.rtmp.RTMPConnection;
1818
import org.red5.server.net.rtmp.codec.RTMPMinaProtocolDecoder;
@@ -35,7 +35,7 @@ public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) th
3535
// get the connection from the session
3636
String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
3737
log.trace("Session id: {}", sessionId);
38-
RTMPConnection conn = (RTMPConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId);
38+
RTMPConnection conn = (RTMPConnection) RTMPClientConnManager.getInstance().getConnectionBySessionId(sessionId);
3939
Red5.setConnectionLocal(conn);
4040
byte[] arr = new byte[in.remaining()];
4141
in.get(arr);
@@ -83,7 +83,7 @@ public void encode(IoSession session, Object message, ProtocolEncoderOutput out)
8383
// get the connection from the session
8484
String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
8585
log.trace("Session id: {}", sessionId);
86-
RTMPConnection conn = (RTMPConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId);
86+
RTMPConnection conn = (RTMPConnection) RTMPClientConnManager.getInstance().getConnectionBySessionId(sessionId);
8787
if (conn != null) {
8888
Red5.setConnectionLocal(conn);
8989
final Semaphore lock = conn.getEncoderLock();

‎client/src/main/java/org/red5/client/net/rtmpe/RTMPEIoFilter.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
import org.apache.mina.filter.codec.ProtocolCodecFilter;
1919
import org.red5.client.net.rtmp.BaseRTMPClientHandler;
2020
import org.red5.client.net.rtmp.OutboundHandshake;
21-
import org.red5.client.net.rtmp.RTMPConnManager;
21+
import org.red5.client.net.rtmp.RTMPClientConnManager;
2222
import org.red5.client.net.rtmp.codec.RTMPMinaCodecFactory;
2323
import org.red5.server.net.rtmp.RTMPConnection;
2424
import org.red5.server.net.rtmp.RTMPMinaConnection;
@@ -43,7 +43,7 @@ public class RTMPEIoFilter extends IoFilterAdapter {
4343
public void messageReceived(NextFilter nextFilter, IoSession session, Object obj) throws Exception {
4444
String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
4545
log.trace("Session id: {}", sessionId);
46-
RTMPMinaConnection conn = (RTMPMinaConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId);
46+
RTMPMinaConnection conn = (RTMPMinaConnection) RTMPClientConnManager.getInstance().getConnectionBySessionId(sessionId);
4747
if (conn == null) {
4848
throw new Exception("Receive on unavailable connection - session id: " + sessionId);
4949
}

‎client/src/main/java/org/red5/client/net/rtmps/RTMPTSClientConnector.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
import org.apache.http.util.EntityUtils;
1919
import org.apache.mina.core.buffer.IoBuffer;
2020
import org.red5.client.net.rtmp.OutboundHandshake;
21-
import org.red5.client.net.rtmp.RTMPConnManager;
21+
import org.red5.client.net.rtmp.RTMPClientConnManager;
2222
import org.red5.client.net.rtmpt.RTMPTClientConnection;
2323
import org.red5.client.net.rtmpt.RTMPTClientConnector;
2424
import org.red5.server.api.Red5;
@@ -134,7 +134,7 @@ private RTMPTClientConnection openConnection() throws IOException {
134134
sessionId = responseStr.substring(0, responseStr.length() - 1);
135135
log.debug("Got an id {}", sessionId);
136136
// create a new connection
137-
conn = (RTMPTClientConnection) RTMPConnManager.getInstance().createConnection(RTMPTClientConnection.class, sessionId);
137+
conn = (RTMPTClientConnection) RTMPClientConnManager.getInstance().createConnection(RTMPTClientConnection.class, sessionId);
138138
log.debug("Got session id {} from connection", conn.getSessionId());
139139
// client state
140140
conn.setHandler(client);

‎client/src/main/java/org/red5/client/net/rtmpt/RTMPTClientConnector.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
import org.apache.http.util.EntityUtils;
2424
import org.apache.mina.core.buffer.IoBuffer;
2525
import org.red5.client.net.rtmp.OutboundHandshake;
26-
import org.red5.client.net.rtmp.RTMPConnManager;
26+
import org.red5.client.net.rtmp.RTMPClientConnManager;
2727
import org.red5.server.api.Red5;
2828
import org.red5.server.net.rtmp.RTMPConnection;
2929
import org.red5.server.net.rtmp.codec.RTMP;
@@ -170,7 +170,7 @@ private RTMPTClientConnection openConnection() throws IOException {
170170
sessionId = responseStr.substring(0, responseStr.length() - 1);
171171
log.debug("Got an id {}", sessionId);
172172
// create a new connection
173-
conn = (RTMPTClientConnection) RTMPConnManager.getInstance().createConnection(RTMPTClientConnection.class, sessionId);
173+
conn = (RTMPTClientConnection) RTMPClientConnManager.getInstance().createConnection(RTMPTClientConnection.class, sessionId);
174174
log.debug("Got session id {} from connection", conn.getSessionId());
175175
// client state
176176
conn.setHandler(client);

‎common/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
<parent>
44
<groupId>org.red5</groupId>
55
<artifactId>red5-parent</artifactId>
6-
<version>1.3.18</version>
6+
<version>1.3.19</version>
77
</parent>
88
<modelVersion>4.0.0</modelVersion>
99
<artifactId>red5-server-common</artifactId>
@@ -124,7 +124,7 @@
124124
<dependency>
125125
<groupId>net.engio</groupId>
126126
<artifactId>mbassador</artifactId>
127-
<version>1.3.18</version>
127+
<version>1.3.19</version>
128128
</dependency> -->
129129
<dependency>
130130
<groupId>junit</groupId>

0 commit comments

Comments
 (0)