I want to have an integrated performance test environment for both HTTP and RTMP. For HTTP,
JMeter is a well known tool, and has a lot of good features. But the problem is that it does not have RTMP sampler.
So I create the RTMP sampler for JMeter. I create the proxy methods using Java Sampler, and from those methods RTMPclient methods are invoked.
I use
Red5 to create RTMPclient. There is a good
tutorial for customizing the RTMPclient.
public class RTMPJavaSampler extends AbstractJavaSamplerClient {
private static HashMap<String, MyRTMPClient> clientMap = new HashMap<String, MyRTMPClient>();
public SampleResult runTest(JavaSamplerContext ctx) {
JMeterVariables vars = JMeterContextService.getContext().getVariables();
SampleResult sampleResult = new SampleResult();
boolean ret = false;
if (ctx.getParameter("MethodName").equals("connect")){
ret = connect(ctx.getParameter("SessionId"), ctx.getParameter("HostIp"), ctx.getParameter("AppName"), ctx.getParameter("Arg1"), ctx.getParameter("Arg2"));
} else if (ctx.getParameter("MethodName").equals("disconnect")){
ret = disconnect(ctx.getParameter("SessionId"));
}
if (ret) {
getLogger().info("Invoke " + ctx.getParameter("MethodName") + " done." );
sampleResult.setResponseData("test success".getBytes());
sampleResult.setDataType(SampleResult.TEXT);
sampleResult.setSuccessful(true);
} else {
getLogger().error("Failed to invoke " + ctx.getParameter("MethodName") + ".");
sampleResult.setResponseData("test failed".getBytes());
sampleResult.setDataType(SampleResult.TEXT);
sampleResult.setSuccessful(false);
}
sampleResult.setResponseCodeOK();
sampleResult.setResponseMessageOK();
return sampleResult;
}
@Override
public Arguments getDefaultParameters() {
Arguments params = new Arguments();
params.addArgument("MethodName", "");
params.addArgument("SessionId", "");
params.addArgument("HostIp", "");
params.addArgument("AppName", "");
params.addArgument("Arg1","");
params.addArgument("Arg2","");
return params;
}
public boolean connect(String sessionId, String hostIp, String appName, String arg1, String arg2){
MyRTMPClient client = new MyRTMPClient();
client.connect(hostIp, appName, arg1, arg2);
clientMap.put(sessionId, client);
while(client.getConnected()==null);
return (client.getConnected());
}
public boolean disconnect(String sessionId){
MyRTMPClient client = clientMap.remove(sessionId);
client.disconnect();
return true;
}
}
public class MyRTMPClient extends RTMPClient implements IPendingServiceCallback, INetStreamEventHandler, IEventDispatcher, ClientExceptionHandler {
private static final Logger logger = LoggingManager.getLoggerForClass();
private Boolean connected;
public MyRTMPClient(){
setServiceProvider(this);
setExceptionHandler(this);
setStreamEventDispatcher(this);
}
public void resultReceived(IPendingServiceCall call) {
Object result = call.getResult();
if ("connect".equals(call.getServiceMethodName())) {
if (result instanceof ObjectMap){
@SuppressWarnings("unchecked")
String code = ((ObjectMap<String, String>) result).get("code");
if (StatusCodes.NC_CONNECT_SUCCESS.equals(code)){
logger.debug("Connection Success.");
connected=true;
} else {
logger.error("resultsReceived: " + code);
connected=false;
}
}
} else if ("createStream".equals(call.getServiceMethodName())){
if (result instanceof Integer) {
Integer streamIdInt = (Integer) result;
// onStreamCreated(streamIdInt);
} else {
logger.error("Unexpected response for createStream " + result.toString());
}
} else {
logger.info(call.getServiceMethodName());
}
}
public void connect(String host, String app, String arg1, String arg2){
Map<String, Object> defParams = makeDefaultConnectionParams(host, 1935, app);
HashMap<String, Object>authParams = new HashMap<String, Object>();
authParams.put("arg1", arg1);
authParams.put("arg2", arg2);
Object[] params = new Object[] {authParams};
connect(host, 1935, defParams, this, params);
logger.info("Connect done: " + app);
}
public void onStatus(Object obj){
@SuppressWarnings("unchecked")
ObjectMap<String, String> map = (ObjectMap<String, String>) obj;
if ("status".equals(map.get("level"))){
logger.debug("onStatus: " + obj.toString());
} else {
logger.info("onStatus: " + obj.toString());
}
}
public void onMetaData(Object obj){
logger.info("onMetaData: " + obj.toString());
}
public void onStreamEvent(Notify notify) {
logger.info("onStreamEvent");
}
public void dispatchEvent(IEvent event) {
if (!(event instanceof IRTMPEvent)) {
logger.debug("skipping non rtmp event: " + event);
return;
}
IRTMPEvent rtmpEvent = (IRTMPEvent) event;
if (rtmpEvent instanceof VideoData) {
} else if (rtmpEvent instanceof AudioData) {
} else if (rtmpEvent instanceof Notify){
}
}
public Boolean getConnected() {
return connected;
}
}
