Использование RTMPClient

Многопользовательские риал-тайм приложения не всегда легко отлаживать. Не так уж редко бывает, что нужно реализовать довольно сложный функционал на сервере, который будет запускаться при определенных действиях на клиентах. Причем не на одном клиенте, а на двух-трех. То есть, в каждом цикле отладки нужно запустить трех клиентов, залогиниться в них тремя разными пользователями, совершить некие действия по-очереди в этих клиентах, и только тогда запуститься нужный метод на сервере.

Понятно, что такая отладка крайне утомительна и не эффективна. Надо решать эту проблему. Первым делом в голову приходит мысль о юнит-тестах. Верно, юнит-тесты весьма спасают, и значительную долю функционала можно реализовать, опираясь на них. Значительную, но не всю. Все-такие какую-то часть поведения нужно отлаживать в условиях взаимодействия сервера с клиентами.

Следующая мысль приходит в голову -- сократить действия в самих клиентах, чтобы быстрее воспроизвести нужную ситуацию. Это тоже помогает, однако это не всегда возможно и не всегда желательно.

И тут мы приходим к клиентам ботам. И вот это очень мощная штука. Для релиз-версии проекта они, может, и не нужны (а бывают и нужны), но в разработке и отладке весьма помогают.

До сих пор таких ботов-клиентов я писал на флэше, ибо только флэш может быть клиентом для RTMP соединения. Но оказалось, что это неправда. В составе Red5 есть класс org.red5.server.net.rtmp.RTMPClient, из которого тож можно извлечь некоторую пользу. С его помощью можно создать бот-клиента как консольную утилиту, а не как флэш приложение. Ну а консольную утилиту гораздо проще будет встроить в какое-нибудь автоматическое тестирование. И тут уже можно думать не только о юнит-тестах, но и о тестах интеграционных. Что было бы весьма неплохо.

Ну что ж, от слов к делу. Создадим в проекте модуль и для него примерно такой пом (фрагмент):

<artifactId>botClient</artifactId>
<groupId>com.company.project</groupId>
<version>1.0.0-SNAPSHOT</version>
<packaging>jar</packaging>

<build>
	<plugins>
		<plugin>
			<groupId>org.apache.maven.plugins</groupId>
			<artifactId>maven-compiler-plugin</artifactId>
			<version>2.3.2</version>
			<configuration>
			</configuration>
		</plugin>
	</plugins>
</build>

Накидаем кое-каких нужных зависимостей

<dependency>
	<groupId>org.red5</groupId>
	<artifactId>red5</artifactId>
	<version>0.9.0</version>
</dependency>

<dependency>
	<groupId>org.apache.mina</groupId>
	<artifactId>mina-core</artifactId>
	<version>2.0.0-RC1</version>
	<type>jar</type>
</dependency>

<dependency>
	<artifactId>slf4j-api</artifactId>
	<groupId>org.slf4j</groupId>
	<version>1.5.11</version>
</dependency>

<dependency>
	<artifactId>slf4j-log4j12</artifactId>
	<groupId>org.slf4j</groupId>
	<version>1.5.11</version>
</dependency>

Что касается org.red5:red5, то этот артефакт я устанавливал вручную, взяв для него red5.jar из дистрибутива Red5. Остальные артефакты берутся из стандартных репозиториев.

Но этого мало. Red5 тянет за собой еще кучу зависимостей, которые тоже придется добавить в пом

<dependency>
	<groupId>bouncycastle</groupId>
	<artifactId>bcprov-jdk16</artifactId>
	<version>140</version>
</dependency>

<dependency>
	<groupId>net.sf.ehcache</groupId>
	<artifactId>ehcache</artifactId>
	<version>1.6.2</version>
</dependency>

<dependency>
	<groupId>commons-beanutils</groupId>
	<artifactId>commons-beanutils</artifactId>
	<version>1.8.2</version>
</dependency>

<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-context</artifactId>
	<version>2.5.6</version>
</dependency>

<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-core</artifactId>
	<version>2.5.6</version>
</dependency>

Ок, с пом разобрались. Теперь нужен главный класс модуля, примерно такой:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BotClientApp
{
	static private final Logger log = LoggerFactory.getLogger(BotClientApp.class);

	static private final String host = "localhost";
	static private final int port = 1935;
	static private final String app = "your-app";

	static public void main(String[] args)
	{
		log.info("BotClientApp started");

		TestPrivateChat testPrivateChat = new TestPrivateChat();
		testPrivateChat.run(host, port, app);
	}
}

Подразумеваем, что у нас будут разные тесты. Тут будет использоваться TestPrivateChat. Но потом добавятся StressTest, TestSomeRareFeature, TestSomeOtherComplexBehavior и т.д.

TestPrivateChat должен создать нужное количество ботов, сказать им, куда коннектиться, как логиниться и чего потом делать.

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TestPrivateChat
{
	private final Logger log = LoggerFactory.getLogger(getClass());

	private BaseClient user1;

	private BaseClient user2;

	public void run(String host, int port, String app)
	{
		log.info("run");

		user1 = new BaseClient("Bot", "pass");
		user1.connect(host,port, app);

		user2 = new BaseClient("Bill", "pass");
		user2.connect(host, port, app);
		
		// do something with bots
	}
}

Ну и самое интересно, это BaseClient -- базовый класс для бота. Наследуем его от org.red5.server.net.rtmp.RTMPClient, и кое чего нужно будет имплеменить, чтобы общаться к RTMP сервером.

public class BaseClient extends RTMPClient implements IPendingServiceCallback, ClientExceptionHandler

Сохраним данные для авторизации и создадим индивидуальный логгер для каждого бота

Logger log;

String name;
String pass;

public BaseClient(String name, String pass)
{
	log = LoggerFactory.getLogger("Client_" + name);
	
	this.name = name;
	this.pass = pass;

	setServiceProvider(this);
	setExceptionHandler(this);
}

Соедиенение с сервером

@Override
public void connect(String host, int port, String app)
{
	log.info("connect {} {} {}", new Object[]{host, port, app});
	super.connect(host, port, app, new ConnectCallback(this));
}

Чтобы обработать успешное соединение и чего-то делать дальше, нам понадобится ConnectCallback (а при не успешном соединении выскочит исключение, и это нам вполне подходит).

class ConnectCallback implements IPendingServiceCallback
{
	BaseClient parent;

	public ConnectCallback(BaseClient parent)
	{
		this.parent = parent;
	}

	public void resultReceived(IPendingServiceCall iPendingServiceCall)
	{
		parent.log.info("onConnect");
		parent.login();
	}
}

Теперь можно логиниться

public void login()
{
	invoke("login", new Object[]{name, pass}, new LoginResultCallback(this));
}

Для обработки результата логина нужен еще один callback

class LoginResultCallback implements IPendingServiceCallback
{
	BaseClient parent;

	public LoginResultCallback(BaseClient parent)
	{
		this.parent = parent;
	}

	public void resultReceived(IPendingServiceCall serviceCall)
	{
		@SuppressWarnings("unchecked")
		Map<String, Object> data = (Map<String, Object>) serviceCall.getResult();
		UserVO user = UserVO.createFromMap(data);
		parent.log.info("onLoginResult " + user);

		parent.selfUser = user;
	}
}

Я сейчас работаю с Wowza, поэтому данные на сервер отправляю в виде Object[], а получаю в виде HashMap<String, Object>. Ибо Wowza не до конца поддерживает AMF сериализацию, в частности, не поддерживает registerClassAlias. Ну а если работать с FMS или Red5, то можно отправлять и получать типизированные объекты.

Ну вот, наш бот соединился с сервером, авторизовался, и может чего-нибудь делать. Может вызывать на сервере какие-нибудь методы и передавать данные:

public void sendMessage(Map<String, Object> msg)
{
	invoke("sendMessage", new Object[]{msg}, this);
}

Или сервер может вызывать методы на клиенте и передавать данные:

@SuppressWarnings({"UnusedDeclaration"})
public void newMessage(Object data)
{
	@SuppressWarnings("unchecked")
	Map<String, Object> msg = (Map<String, Object>) data;
	log.info("newMessage " + msg.get("senderName") + " " + msg.get("color") + " " + msg.get("content"));
}

Да @SuppressWarnings понадобятся в изобилии )

Так, все работает, но вот только в логах у нас не все хорошо -- много шума. Классы Red5 генерируют много своих логов, и становиться трудно отслеживать свои. Нужно примерить фильтрацию. А для этого воспользуемся StringMatchFilter.

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">

<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">

	<appender name="console" class="org.apache.log4j.ConsoleAppender">

		<param name="Target" value="System.out"/>

		<layout class="org.apache.log4j.PatternLayout">
			<param name="ConversionPattern" value="%-5p %c{1} - %m%n"/>
		</layout>

		<filter class="org.apache.log4j.varia.StringMatchFilter">
			<param name="StringToMatch" value="ChunkSize is not implemented yet"/>
			<param name="AcceptOnMatch" value="false"/>
		</filter>
		
		<filter class="org.apache.log4j.varia.StringMatchFilter">
			<param name="StringToMatch" value="Stream doesn't exist any longer"/>
			<param name="AcceptOnMatch" value="false"/>
		</filter>

	</appender>

	<root>
		<priority value="info"/>
		<appender-ref ref="console"/>
	</root>

</log4j:configuration>

И вот теперь все клево )