Бесплатный реалтайм список онлайн юзеров (Parse.com + Pubnub)

/*global Parse:false, $:false, jQuery:false */

// Importas
var _ = require('underscore'); // jshint ignore:line
var moment = require('moment'); // jshint ignore:line

// Constants
var sessionObjName = "IMH_Session";
var sessionLifetimeSec = 13;

var channelName = "events";
var publishKey = "pub-c-6271f363-519a-432d-9059-e65a7203ce0e",
    subscribeKey = "sub-c-a3d06db8-410b-11e5-8bf2-0619f8945a4f",
    httpRequestUrl = 'http://pubsub.pubnub.com/publish/' + publishKey + '/' + subscribeKey + '/0/' + channelName + '/0/';


// Utils
function Log(obj, tag) {
    "use strict";

    var loggingString = "Cloud_code: ";
    if (tag != null) { // jshint ignore:line
        loggingString += "[" + tag + "] ";
    }
    loggingString += JSON.stringify(obj) + "\n";

    console.log(loggingString); // jshint ignore:line
}

function GetNow() {
    "use strict";
    return moment.utc();
}

// Supporting
var baseSession = {udid: "", loginedAt: GetNow(), aliveTo: GetNow()};

var errorHandler = function(error) {
    "use strict";
    Log(error.message, "error");
};

function DeleteSession(obj) {
	obj.set("loginedAt", obj.get("aliveTo"));
	SendEvent(obj); 
    obj.destroy();
}

function DeleteDeadSessions() {
    "use strict";

    var query = new Parse.Query(sessionObjName); // jshint ignore:line
    var promise = query.lessThanOrEqualTo("aliveTo", GetNow().toDate())
        .each(function(obj)
         {
             Log(obj, "Delete dead session");
			 DeleteSession(obj);
         }
    );
	return promise;
}

function NewSession(udid) {
    "use strict";

    var session = _.clone(baseSession);
    session.udid = udid;
    session.loginedAt = GetNow();
    session.aliveTo = GetNow().add({seconds: sessionLifetimeSec});

    return session;
}

function GetSessionQuery() {
    "use strict";
    var objConstructor = Parse.Object.extend(sessionObjName); // jshint ignore:line
	var query = new Parse.Query(objConstructor);
    //query.select("udid", "loginedAt", "aliveTo"); //not work for some reason
	return query;
}

function IsUserOnline(udid, onUserOnlineHanlder, onUserOfflineHanlder, onError) {
	"use strict";

	var userAlive = false;
	var query = GetSessionQuery();
	query.equalTo("udid", udid).greaterThanOrEqualTo("aliveTo", GetNow().toDate());
	query.find({
		success: function(result)
		{
			if (result.length == 0) {
				onUserOfflineHanlder();
			}
			else {
				onUserOnlineHanlder(result);
			}
		},
		error: onError
	});
}

function NewParseSession(session) {
    "use strict";

    var objConstructor = Parse.Object.extend(sessionObjName); // jshint ignore:line
    var obj = new objConstructor();

    obj.set({
        udid: session.udid,
        loginedAt: session.loginedAt.toDate(),
        aliveTo: session.aliveTo.toDate()
        }
    );

    return obj;
}

function SendEvent(session) {
    "use strict";

    Parse.Cloud.httpRequest({ // jshint ignore:line
        url: httpRequestUrl + JSON.stringify(session),

        success: function(httpResponse) {},
        error: function(httpResponse) {
            Log('Request failed with response code ' + httpResponse.status);
        }
    });
}

// API functions
var API_GetNow = function(request, response) {
    "use strict";

	var onUserOnline = function(result) {
		response.success( GetNow().toDate() );
	};

	var onUserOffline = function(error) {
		response.error(error);
	};

	var onError = function(error) {
		response.error(error);
	};

	IsUserOnline(request.params.udid, onUserOnline, onUserOffline, onError);
};

var API_GetOnlineUsers = function(request, response) {
    "use strict";

	var onUserOnline = function(result) {
		var query = GetSessionQuery()
		    .addDescending("aliveTo");
		query.find({
			success: function(result)
			{
				response.success( JSON.stringify(result) );
			},
			error: errorHandler
		});
	};

	var onUserOffline = function(error) {
		response.error(error);
	};

	var onError = function(error) {
		response.error(error);
	};

    DeleteDeadSessions().always( function() {
	    IsUserOnline(request.params.udid, onUserOnline, onUserOffline, onError);
    });
};

var API_Login = function(request, response) {
    "use strict";

    var userUdid = request.params.udid;
    var session = NewSession(userUdid);
    var parseObject = NewParseSession(session);

	Parse.Cloud.run("Logout", {udid: userUdid}).always( function() {
		parseObject.save(null, {
			success: function(obj) {
				Log(obj, "Login:save");
				response.success( JSON.stringify(parseObject) );
			},
			error: function(error) {
				errorHandler(error);
				response.error(error);
			}
		});
	});
};

var API_Logout = function(request, response) {
    "use strict";

    var userUdid = request.params.udid;
    var query = GetSessionQuery()
        .equalTo("udid", userUdid);

    query.each( function(obj) {
        Log(obj, "Logout:destroy");
		DeleteSession(obj);
    }).done( function() {response.success();} );
};


// Bindings
Parse.Cloud.afterSave(sessionObjName, function(request) { // jshint ignore:line
    "use strict";

	SendEvent(request.object);
});

// API definitions
Parse.Cloud.define("GetNow", API_GetNow); // jshint ignore:line

Parse.Cloud.define("GetOnlineUsers", API_GetOnlineUsers); // jshint ignore:line

Parse.Cloud.define("Login", API_Login); // jshint ignore:line
Parse.Cloud.define("Logout", API_Logout); // jshint ignore:line

public class PubnubChannel extends Channel {
	static private final String CHANNEL_NAME = "events";
	static private final String SUBSCRIBE_KEY = "sub-c-a3d06db8-410b-11e5-8bf2-0619f8945a4f";

	Pubnub pubnub = new Pubnub("", SUBSCRIBE_KEY);
	Callback pubnubCallback = new Callback() {
		@Override
		public void connectCallback(String channel, Object message) {
			if (listener != null) {
				listener.onConnect(channel, "Connected: " + message.toString());
			}
		}

		@Override
		public void disconnectCallback(String channel, Object message) {
			if (listener != null) {
				listener.onDisconnect(channel, "Disconnected: " + message.toString());
			}
		}

		@Override
		public void reconnectCallback(String channel, Object message) {
			if (listener != null) {
				listener.onReconnect(channel, "Reconnected: " + message.toString());
			}
		}

		@Override
		public void successCallback(String channel, Object message, String timetoken) {
			if (listener != null) {
				listener.onMessageRecieve(channel, message.toString(), timetoken);
			}
		}

		@Override
		public void errorCallback(String channel, PubnubError error) {
			if (listener != null) {
				listener.onErrorOccur(channel, "Error occured: " + error.toString());
			}
		}
	};

	public PubnubChannel() {
		setName(CHANNEL_NAME);
	}

	@Override
	public void subscribe() throws ChannelException {
		try {
			pubnub.subscribe(CHANNEL_NAME, pubnubCallback);
		} catch (PubnubException e) {
			e.printStackTrace();
			throw new ChannelException(ChannelException.CONNECT_ERROR, e);
		}
	}

	@Override
	public void unsubscribe() {
		pubnub.unsubscribeAll();
	}
}

public class ServerChannel {
	Logger l = LoggerFactory.getLogger(ServerChannel.class);

	JsonParser jsonParser;
	Channel serverChannel;

	ServerChannel.EventListener listener;
	private final Channel.EventListener listenerAdapter = new Channel.EventListener() {
		@Override
		public void onConnect(String channel, String greeting) {

		}

		@Override
		public void onDisconnect(String channel, String reason) {
			if (listener != null) {
				listener.onDisconnect(reason);
			}
		}

		@Override
		public void onReconnect(String channel, String reason) {

		}

		@Override
		public void onMessageRecieve(String channel, String message, String timetoken) {
			if (listener != null) {
				ServerChannel.this.onMessageRecieve(message, timetoken);
			}
		}

		@Override
		public void onErrorOccur(String channel, String error) {
			l.warn(String.format("%s : [error] %s", channel, error));
			if (listener != null) {
				ServerChannel.this.unsubscribe();
			}
		}
	};

	public ServerChannel(Channel serverChannel, JsonParser jsonParser) {
		this.serverChannel = serverChannel;
		this.jsonParser = jsonParser;
	}

	public final void setListener(@NonNull ServerChannel.EventListener listener) {
		this.listener = listener;
	}

	public final void clearListener() {
		listener = null;
	}

	public final void subscribe() throws ChannelException {
		try {
			serverChannel.setListener(listenerAdapter);
			serverChannel.subscribe();
		} catch (ChannelException e) {
			e.printStackTrace();
			serverChannel.clearListener();
			throw e;
		}
	}
	public final void unsubscribe() {
		serverChannel.unsubscribe();
		serverChannel.clearListener();
	}

	public void onMessageRecieve(String userJson, String timetoken) {
		DyingUser dyingUser = jsonParser.fromJson(userJson, DyingUser.class);
		if (dyingUser != null) {
			if (dyingUser.isAlive()) {
				listener.onUserLogin(dyingUser);
			} else {
				listener.onUserLogout(dyingUser);
			}
		}
	}

	public interface EventListener {
		void onDisconnect(String reason);
		void onUserLogin(DyingUser dyingUser);
		void onUserLogout(DyingUser dyingUser);
	}
}

public class AuthApi extends Api {
	static final String
			API_Login = "Login",
			API_Logout = "Logout";

	@Inject
	public AuthApi(JsonParser parser) {
		super(parser);
	}

	public DyingUser login(@NonNull final String udid) throws ApiException {
		DyingUser dyingUser;
		try {
			String jsonObject = ParseCloud.callFunction(API_Login, constructRequestForUser(udid));
			dyingUser = parser.fromJson(jsonObject, DyingUser.class);
		} catch (ParseException e) {
			e.printStackTrace();
			throw new ApiException(ApiException.LOGIN_ERROR, e);
		}
		return dyingUser;
	}

	public void logout(@NonNull final DyingUser dyingUser) {
		try {
			ParseCloud.callFunction(API_Logout, constructRequestForUser(dyingUser.getUdid()));
		} catch (ParseException e) {
			e.printStackTrace();
		}
	}
}

public class UserApi extends Api {
	static final String
			API_GetOnlineUsers = "GetOnlineUsers";

	@Inject
	public UserApi(JsonParser parser) {
		super(parser);
	}

	public final ArrayList<DyingUser> getOnlineUsers(@NonNull final DyingUser dyingUser) throws ApiException {
		ArrayList<DyingUser> users;
		try {
			String jsonUsers = ParseCloud.callFunction(API_GetOnlineUsers, constructRequestForUser(dyingUser.getUdid()));
			users = parser.fromJson(jsonUsers, new TypeToken<List<DyingUser>>(){}.getType());
		} catch (ParseException e) {
			e.printStackTrace();
			throw new ApiException(ApiException.GET_USERS_ERROR, e);
		}
		return users;
	}
}

abstract class Api {
	final JsonParser parser;

	Api(JsonParser parser) {
		this.parser = parser;
	}

	protected Map<String, ?> constructRequestForUser(@NonNull final String udid)
	{
		Map<String, String> result = new HashMap<>();
		result.put("udid", udid);
		return result;
	}
}

public class TimeTicker extends Listenable<TimeTicker.EventListener> {
	private static final long TICKING_PERIOD_MS_DEFAULT = 1000;
	private static final boolean DO_INSTANT_TICK_ON_START_DEFAULT = true;
	long tickingPeriodMs;
	boolean doInstantTickOnStart;

	final Handler uiHandler = new Handler(Looper.getMainLooper());
	final Timer tickingTimer = new Timer();
	TimerTask tickingTask;

	public TimeTicker() {
		this(DO_INSTANT_TICK_ON_START_DEFAULT);
	}

	public TimeTicker(boolean doInstantTickOnStart) {
		this.doInstantTickOnStart = doInstantTickOnStart;
		setTickingPeriodMs(TICKING_PERIOD_MS_DEFAULT);
	}

	public void setTickingPeriodMs(final long tickingPeriodMs) {
		this.tickingPeriodMs = tickingPeriodMs;
	}

	public synchronized void start() {
		if (tickingTask != null) {
			stop();
		}

		tickingTask = new TimerTask() {
			@Override
			public void run() {
				uiHandler.post(new Runnable() {
					@Override
					public void run() {
						forEachListener(new ListenerExecutor<TimeTicker.EventListener>() {
							@Override
							public void run() {
								getListener().onSecondTick();
							}
						});
					}
				});
			}
		};

		long delay = (doInstantTickOnStart) ? 0 : tickingPeriodMs;
		tickingTimer.scheduleAtFixedRate(tickingTask, delay, tickingPeriodMs);
	}

	public synchronized void stop() {
		if (tickingTask != null) {
			tickingTask.cancel();
		}
		tickingTask = null;
		tickingTimer.purge();
	}

	public interface EventListener extends Listenable.EventListener {
		void onSecondTick();
	}

	public interface Owner {
		TimeTicker getTimeTicker();
	}
}

public class TemporarySet<TItem> extends Listenable<TemporarySet.EventListener> implements Resumable {
	protected final SortedSet<TemporaryElement<TItem>> sortedElementsSet = new TreeSet<>();
	protected final List<TItem> list = new ArrayList<>();

	protected final Timer timer = new Timer();
	protected TimerTask timerTask = null;
	protected TemporaryElement<TItem> nextElementToDie = null;

	boolean isResumed = false;

	public TemporarySet() {
		notifier = new TemporarySet.EventListener() {
			@Override
			public void onCleared() {
				for (TemporarySet.EventListener listener : getListenersSet()) {
					listener.onCleared();
				}
			}

			@Override
			public void onAdded(Object item) {
				for (TemporarySet.EventListener listener : getListenersSet()) {
					listener.onAdded(item);
				}
			}

			@Override
			public void onRemoved(Object item) {
				for (TemporarySet.EventListener listener : getListenersSet()) {
					listener.onRemoved(item);
				}
			}
		};
	}

	public boolean add(TItem object, DateTime deathTime) {
		TemporaryElement<TItem> element = new TemporaryElement<>(object, deathTime);
		return _add(element);
	}

	public boolean remove(TItem object) {
		TemporaryElement<TItem> element = new TemporaryElement<>(object);
		return _remove(element);
	}

	public void clear() {
		_clear();
	}

	public final List<TItem> asReadonlyList() {
		return Collections.unmodifiableList(list);
	}

	private synchronized void _clear() {
		cancelNextDeath();
		list.clear();
		sortedElementsSet.clear();

		notifier.onCleared();
	}


	private synchronized boolean _add(TemporaryElement<TItem> insertingElement) {
		boolean wasInserted = _insertElementUnique(insertingElement);

		if (wasInserted) {
			if (nextElementToDie != null &&
					nextElementToDie.deathTime.isAfter(insertingElement.deathTime)) {
				cancelNextDeath();
			}

			if (nextElementToDie == null) {
				openNextDeath();
			}

			notifier.onAdded(insertingElement.object);
		}

		return wasInserted;
	}

	private synchronized boolean _remove(TemporaryElement<TItem> deletingElement) {
		boolean wasDeleted = _deleteElementByObject(deletingElement);

		if (wasDeleted) {
			if (nextElementToDie.equals(deletingElement)) {
				cancelNextDeath();
				openNextDeath();
			}

			notifier.onRemoved(deletingElement.object);
		}

		return wasDeleted;
	}

	private synchronized void openNextDeath() {
		cancelNextDeath();
		if (sortedElementsSet.size() != 0) {
			nextElementToDie = sortedElementsSet.first();
			timerTask = new TimerTask() {
				@Override
				public void run() {
					_remove(nextElementToDie);
				}
			};

			DateTime now = new DateTime();
			Duration duration = TimeUtils.GetNonNegativeDuration(now, nextElementToDie.deathTime);

			timer.schedule(timerTask, duration.getMillis());
		}
	}

	private synchronized void cancelNextDeath() {
		if (timerTask != null) {
			timerTask.cancel();
		}
		timer.purge();
		nextElementToDie = null;
		timerTask = null;
	}

	private synchronized Iterator<TemporaryElement<TItem>> findElement(TemporaryElement<TItem> searchingElement) {
		Iterator<TemporaryElement<TItem>> resultIterator = null;
		for (Iterator<TemporaryElement<TItem>> iterator = sortedElementsSet.iterator(); iterator.hasNext() && resultIterator == null;) {
			if (iterator.next().equals(searchingElement)) {
				resultIterator = iterator;
			}
		}
		return resultIterator;
	}

	private synchronized boolean _insertElementUnique(TemporaryElement<TItem> element) {
		boolean wasInserted = false;

		Iterator<TemporaryElement<TItem>> iterator = findElement(element);
		if (iterator == null) {
			wasInserted = true;
			sortedElementsSet.add(element);
			list.add(element.object);
		}

		return wasInserted;
	}

	private synchronized boolean _deleteElementByObject(TemporaryElement<TItem> element) {
		boolean wasDeleted = false;

		Iterator<TemporaryElement<TItem>> iterator = findElement(element);
		if (iterator != null) {
			wasDeleted = true;
			iterator.remove();
			list.remove(element.object);
		}

		return wasDeleted;
	}

	@Override
	public void resume() {
		isResumed = true;
		openNextDeath();
	}

	@Override
	public void pause() {
		cancelNextDeath();
		isResumed = false;
	}

	@Override
	public boolean isResumed() {
		return isResumed;
	}

	public interface EventListener extends Listenable.EventListener {
		void onCleared();
		void onAdded(Object item);
		void onRemoved(Object item);
	}
}

class TemporaryElement<T> implements Comparable {
	protected final T object;
	protected final DateTime deathTime;

	public TemporaryElement(@NonNull T object, @NonNull DateTime deathTime) {
		this.deathTime = deathTime;
		this.object = object;
	}

	public TemporaryElement(@NonNull T object) {
		this(object, new DateTime(0));
	}

	@Override
	public boolean equals(Object o) {
		if (this == o) return true;
		if (o == null || getClass() != o.getClass()) return false;

		TemporaryElement<?> that = (TemporaryElement<?>) o;

		return object.equals(that.object);
	}

	@Override
	public int hashCode() {
		return object.hashCode();
	}

	@Override
	public int compareTo(@NonNull Object another) {
		TemporaryElement a = this,
				b = (TemporaryElement) another;

		int datesComparisionResult = a.deathTime.compareTo(b.deathTime);
		int objectsComparisionResult = a.hashCode() - b.hashCode();
		return (datesComparisionResult != 0) ? datesComparisionResult : objectsComparisionResult;
	}
}

Parse.Cloud.afterSave("Foo", function(request) {}); // custom Foo object
Parse.Cloud.afterSave("User", function(request) {}); // custom(!) User object
Parse.Cloud.afterSave(Parse.User, function(request) {}); // Parse.com User object
Parse.Cloud.afterSave(Parse.Session, function(request) {}); // error! can't bind to Parse.Session