Custom module to count how much traffic users have done in a X range of seconds

Mauricio_Antunes · September 8, 2014, 10:23am

I am currently trying to build a module to improve accounting of the streams made by Wowza.

Software versions used to build this module:

Wowza version: 4

JDK: 1.8

Rationale:

I want to build a module that:

Counts how much traffic users have done in a range of seconds e.g 10 (disconnect/stop should count as well)

Available for RTMP, RTP(RTSP) and HTTP (HLS/HDS/DASH/Smooth)

Problems:

I would like to know what is a proper way to count all those protocol in the same module.

What I have so far:

package com.mycompany.wms.module;
import java.util.Timer;
import java.util.TimerTask;
import com.wowza.util.IOPerformanceCounter;
import com.wowza.wms.application.*;
import com.wowza.wms.amf.*;
import com.wowza.wms.client.*;
import com.wowza.wms.httpstreamer.model.IHTTPStreamerSession;
import com.wowza.wms.media.model.MediaCodecInfoAudio;
import com.wowza.wms.media.model.MediaCodecInfoVideo;
import com.wowza.wms.module.*;
import com.wowza.wms.request.*;
import com.wowza.wms.stream.IMediaStream;
import com.wowza.wms.stream.IMediaStreamActionNotify3;
public class AccountingModule extends ModuleBase {
	IApplicationInstance appInstance;
  
	public void onAppStart(IApplicationInstance appInstance) {
		this.appInstance = appInstance;
		String fullname = appInstance.getApplication().getName() + "/"
				+ appInstance.getName();
		getLogger().info("onAppStart: " + fullname);
	}
	public void onAppStop(IApplicationInstance appInstance) {
		String fullname = appInstance.getApplication().getName() + "/"
				+ appInstance.getName();
		getLogger().info("onAppStop: " + fullname);
	}
	public void onConnect(IClient client, RequestFunction function,
			AMFDataList params) {
		getLogger().info("onConnect: " + client.getClientId());
	}
	public void onConnectAccept(IClient client) {
		getLogger().info("onConnectAccept: " + client.getClientId());
	}
	public void onConnectReject(IClient client) {
		getLogger().info("onConnectReject: " + client.getClientId());
	}
	public void onDisconnect(IClient client) {
		getLogger().info("onDisconnect: " + client.getClientId());
	}
	class StreamListener implements IMediaStreamActionNotify3 {
		@Override
		public void onMetaData(IMediaStream stream, AMFPacket metaDataPacket) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onPauseRaw(IMediaStream stream, boolean isPause,
				double location) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onPause(IMediaStream stream, boolean isPause,
				double location) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onPlay(IMediaStream stream, String streamName,
				double playStart, double playLen, int playReset) {
			getLogger().info(
					"onPlay " + stream.getContextStr() +
							" playStart: " + playStart + 
							" playLen: " + playLen +
							" playReset: "+ playReset);
			StreamStats watchdog = new StreamStats();
			IApplicationInstance appInstance;
			try {
				try {
					appInstance = stream.getClient().getAppInstance();
				} catch (Exception ex) {
					appInstance = stream.getRTPStream().getAppInstance();
				}
				watchdog.start();
				appInstance.getProperties().setProperty(streamName, watchdog);
			} catch (Exception ex) {
			}
		}
		@Override
		public void onPublish(IMediaStream stream, String streamName,
				boolean isRecord, boolean isAppend) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onSeek(IMediaStream stream, double location) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onStop(IMediaStream stream) {
			getLogger().info("onStop By: " + stream.getClientId());
			String streamName = stream.getName();
			StreamStats watchdog = (StreamStats) stream.getClient()
					.getAppInstance().getProperties().getProperty(streamName);
			if (watchdog != null)
				watchdog.stop();
		}
		@Override
		public void onUnPublish(IMediaStream stream, String streamName,
				boolean isRecord, boolean isAppend) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onCodecInfoAudio(IMediaStream stream,
				MediaCodecInfoAudio codecInfoAudio) {
			// TODO Auto-generated method stub
		}
		@Override
		public void onCodecInfoVideo(IMediaStream stream,
				MediaCodecInfoVideo codecInfoVideo) {
			// TODO Auto-generated method stub
		}
	}
	@SuppressWarnings("unchecked")
	public void onStreamCreate(IMediaStream stream) {
		getLogger().info("onStreamCreate: " + stream);
		IMediaStreamActionNotify3 actionNotify = new StreamListener();
		WMSProperties props = stream.getProperties();
		synchronized (props) {
			props.put("streamActionNotifier", actionNotify);
		}
		stream.addClientListener(actionNotify);
	}
	public void onStreamDestroy(IMediaStream stream) {
		getLogger().info("onStreamDestroy: " + stream);
		IMediaStreamActionNotify3 actionNotify = null;
		WMSProperties props = stream.getProperties();
		synchronized (props) {
			actionNotify = (IMediaStreamActionNotify3) stream.getProperties()
					.get("streamActionNotifier");
		}
		if (actionNotify != null) {
			stream.removeClientListener(actionNotify);
			getLogger().info("remoteClientListener: " + stream.getSrc());
		}
	}
	public void onHTTPSessionCreate(IHTTPStreamerSession httpSession) {
		String streamName = httpSession.getStreamName();
		getLogger().info("Stream Name: " + streamName);
		StreamStats watchdog = new StreamStats();
		try {
			watchdog.session = httpSession;
			watchdog.start();
			this.appInstance.getProperties().setProperty(
					streamName + httpSession.getSessionId(), watchdog);
		} catch (Exception ex) {
		}
	}
	private class StreamStats {
		public Timer mTimer;
		public TimerTask mTask;
		public IHTTPStreamerSession session;
  	
		public StreamStats() {
			mTask = new TimerTask() {
				public void run() {
					getLogger().info("Run StreamStats: " + session.getStreamName());
					if (session != null) {
						IOPerformanceCounter perf = session
								.getIOPerformanceCounter();	
						getLogger().info(
								"Bytes loaded till now: " + perf.getMessagesOutBytes());
					}
				}
			};
		}
		public void start() {
			if (mTimer == null)
				mTimer = new Timer();
			mTimer.schedule(mTask, 10000, 10000);
			getLogger().info("Start StreamStats");
		}
		public void stop() {
			if (mTimer != null) {
				mTimer.cancel();
				mTimer = null;
				getLogger().info("Stop StreamStats");
			}
		}
	}
}

Matt_Young · September 9, 2014, 5:04am

Hello

One example would be to check out our connection counts http provider.

Thanks

Matt

Mauricio_Antunes · September 9, 2014, 11:06am

Would I be able to count traffic (bytes) with that example if using JWPlayer? (rtmp://[wowza-ipaddress]/…/live.sdp)

Mauricio_Antunes · September 10, 2014, 4:42pm

Hey matt_y my module looks like it now:

package com.mycompany.wms.module;
 
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import com.wowza.util.IOPerformanceCounter;
import com.wowza.wms.amf.AMFPacket;
import com.wowza.wms.application.IApplicationInstance;
import com.wowza.wms.application.WMSProperties;
import com.wowza.wms.httpstreamer.model.IHTTPStreamerSession;
import com.wowza.wms.media.model.MediaCodecInfoAudio;
import com.wowza.wms.media.model.MediaCodecInfoVideo;
import com.wowza.wms.module.ModuleBase;
import com.wowza.wms.stream.IMediaStream;
import com.wowza.wms.stream.IMediaStreamActionNotify3;
 
 
public class AccountingModule extends ModuleBase {
	
	IApplicationInstance appInstance;
	
	public void onAppStart(IApplicationInstance appInstance) {
		this.appInstance = appInstance;
	}
	
	public void onStreamDestroy(IMediaStream stream) {
		getLogger().info("onStreamDestroy");
		
		IMediaStreamActionNotify3 actionNotify = null;
		WMSProperties props = stream.getProperties();
		synchronized (props) {
			actionNotify = (IMediaStreamActionNotify3)stream.getProperties().get("streamActionNotifier");
		}
		if (actionNotify != null) {
			stream.removeClientListener(actionNotify);
			getLogger().info("removeClientListener " + stream.getSrc());
		}
	}
	
	@SuppressWarnings("unchecked")
	public void onStreamCreate(IMediaStream stream) {
		getLogger().info("onStreamCreate");
		
		IMediaStreamActionNotify3 actionNotify = new StreamListener();
		WMSProperties props = stream.getProperties();
		synchronized (props) {
			props.put("streamActionNotifier", actionNotify);
		}
		stream.addClientListener(actionNotify);
	}
	
	class StreamListener implements IMediaStreamActionNotify3{
		@Override
		public void onMetaData(IMediaStream stream, AMFPacket metaDataPacket) {
			// TODO Auto-generated method stub
			
		}
		@Override
		public void onPauseRaw(IMediaStream stream, boolean isPause,
				double location) {
			// TODO Auto-generated method stub
			
		}
		@Override
		public void onPause(IMediaStream arg0, boolean arg1, double arg2) {
			// TODO Auto-generated method stub
			
		}
		@Override
		public void onPlay(IMediaStream stream, String streamName, double playStart, double playLen, int playReset) {
			getLogger().info("onPlay " + streamName);
			StreamStats watchdog = new StreamStats();
			IApplicationInstance appInstance;
			
			try {
				try {
					appInstance = stream.getClient().getAppInstance();
				} catch (Exception ex) {
					appInstance = stream.getRTPStream().getAppInstance();
				}
				
				watchdog.stream = stream;
				watchdog.start();
				appInstance.getProperties().setProperty(streamName, watchdog);
			} catch (Exception ex) {
				
			}
		}
		@Override
		public void onUnPublish(IMediaStream arg0, String arg1, boolean arg2,
				boolean arg3) {
			// TODO Auto-generated method stub
			
		}
		
		@Override
		public void onPublish(IMediaStream stream, String streamName, boolean isRecord, boolean isAppend) {
			// TODO Auto-generated method stub
			
		}
		@Override
		public void onSeek(IMediaStream arg0, double arg1) {
			// TODO Auto-generated method stub
			
		}
		@Override
		public void onStop(IMediaStream stream) {
			getLogger().info("onStop");
			
			String streamName = stream.getName();
			StreamStats watchdog = (StreamStats)stream.getClient().getAppInstance().getProperties().getProperty(streamName);
			if (watchdog != null)
				getLogger().info("Stop Watchdog");
				watchdog.stop();
		}
		@Override
		public void onCodecInfoAudio(IMediaStream stream,
				MediaCodecInfoAudio codecInfoAudio) {
			// TODO Auto-generated method stub
			
		}
		@Override
		public void onCodecInfoVideo(IMediaStream stream,
				MediaCodecInfoVideo codecInfoVideo) {
			// TODO Auto-generated method stub
			
		}
		
	}
	
	public void onHTTPSessionDestroy(IHTTPStreamerSession httpSession) {
		getLogger().info("onHTTPSessionDestroy");
		
		String streamName = httpSession.getStreamName();
		StreamStats watchdog = (StreamStats)this.appInstance.getProperties().getProperty(streamName + httpSession.getSessionId());
		if (watchdog != null)
			getLogger().info("Stop Watchdog");
			watchdog.stop();
	}
	
	public void onHTTPSessionCreate(IHTTPStreamerSession httpSession) {
		getLogger().info("onHTTPSessionCreate");
		
		String streamName = httpSession.getStreamName();
		getLogger().info("Stream Name: " + streamName);
		StreamStats watchdog = new StreamStats();
		
		try {
			watchdog.session = httpSession;
			watchdog.start();
			this.appInstance.getProperties().setProperty(streamName + httpSession.getSessionId(), watchdog);
		} catch (Exception ex) {
		}
	}
 
	private class StreamStats
	{
		public Timer mTimer;
		public TimerTask mTask;
		public IMediaStream stream;
		public IHTTPStreamerSession session;
		
		public String connectionId;
		public long bytesOut;
		public long bytesNow;
		
		Map<String, Long> usersMap = new HashMap<String, Long>();
		
		public StreamStats() {
			mTask = new TimerTask() {
				public void run() {
					getLogger().info("Run StreamStats");
					if (session != null) {
						IOPerformanceCounter perf = session.getIOPerformanceCounter();
						getLogger().info("bytesOut " + perf.getMessagesOutBytes());
						bytesOut = perf.getMessagesOutBytes();
						
						connectionId = session.getSessionId();
					}
					if (stream != null) {
						IOPerformanceCounter perf = stream.getMediaIOPerformance();
						getLogger().info("bytesOut " + perf.getMessagesOutBytes());
					
						connectionId = Long.toString(stream.getClientId());
						bytesOut = perf.getMessagesOutBytes();
					}
					
					if (usersMap.containsKey(connectionId)) {
						bytesNow = bytesOut - usersMap.get(connectionId);
					} else {
						bytesNow = bytesOut - 0;
					}
					usersMap.put(connectionId, bytesOut);
					
					getLogger().info("connectionId " + connectionId);
					getLogger().info("bytesNow " + bytesNow);
				}
			};
		}
		
		public void start() {
			if (mTimer == null)
				mTimer = new Timer();
			mTimer.schedule(mTask, 10000, 10000);
			getLogger().info("Start StreamStats");
			
		}
		
		public void stop() {
			if (mTimer != null) {
				mTimer.cancel();
				mTimer = null;
				getLogger().info("Stop StreamStats");
			}
		}
	}
}

Do you see any problem with this?

I am not sure if this is the best way to uniform RTMP, RTP, HTTP in one module.

Mauricio_Antunes · September 15, 2014, 1:03pm

Now I get what you meant talking about HTTPProvider.

Implementing something like that, how to avoid the HTTP overhead? Is there any way to use that module (example provided in the link) without the HTTP overhead? Like calling from other class? There is the timer (range of seconds) problem too. What are your thoughts?

Topic		Replies	Views
Logging player counts to database Wowza Developer Dojo	2	2145	June 7, 2014
custom module creator people recommendations Wowza Developer Dojo wowza-streaming-server-java-api	2	2095	September 23, 2015
Starting to develop a custom module Wowza Developer Dojo wowza-streaming-server-java-api	4	5803	May 17, 2021
Custom module to stop stream from post variables Wowza Developer Dojo wowza-streaming-server-java-api	8	4201	May 6, 2014
Wowza Custom IMediaWriterActionNotify Wowza Developer Dojo wowza-streaming-server-java-api	2	2348	February 9, 2017

Custom module to count how much traffic users have done in a X range of seconds

Related topics