RTMP sampler for JMeter to test flash with http

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;
    }
}