package com.pubnub.api;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.junit.Test;
import java.util.Hashtable;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.*;
public class PubnubTest {
class SubscribeCallback extends Callback {
private CountDownLatch latch;
private Object response;
public SubscribeCallback(CountDownLatch latch) {
this.latch = latch;
}
public SubscribeCallback() {
}
public Object getResponse() {
return response;
}
@Override
public void successCallback(String channel, Object message) {
response = message;
if (latch != null)
latch.countDown();
}
@Override
public void errorCallback(String channel, PubnubError error) {
response = error;
if (latch != null)
latch.countDown();
}
}
class PublishCallback extends Callback {
private CountDownLatch latch;
private int result = 0;
public int getResult() {
return result;
}
public PublishCallback(CountDownLatch latch) {
this.latch = latch;
}
public PublishCallback() {
}
public void successCallback(String channel, Object message) {
JSONArray jsarr;
try {
jsarr = (JSONArray) message;
result = (Integer) jsarr.get(0);
} catch (JSONException e) {
e.printStackTrace();
}
if (latch != null)
latch.countDown();
}
public void errorCallback(String channel, PubnubError error) {
JSONArray jsarr;
result = 0;
if (latch != null)
latch.countDown();
}
}
class UlsCallback extends Callback {
private CountDownLatch latch;
private JSONObject result = null;
public JSONObject getResult() {
return result;
}
public UlsCallback(CountDownLatch latch) {
this.latch = latch;
}
public UlsCallback() {
}
public void successCallback(String channel, Object message) {
result = (JSONObject) message;
if (latch != null)
latch.countDown();
}
public void errorCallback(String channel, PubnubError error) {
result = null;
if (latch != null)
latch.countDown();
}
}
class HereNowCallback extends Callback {
private int occupancy;
private String[] uuids;
private CountDownLatch latch;
public int getOccupancy() {
return occupancy;
}
public HereNowCallback(CountDownLatch latch) {
this.latch = latch;
}
public HereNowCallback() {
}
@Override
public void successCallback(String channel, Object message) {
JSONObject resp = null;
try {
resp = new JSONObject(message.toString());
} catch (JSONException e1) {
e1.printStackTrace();
}
if (resp != null) {
try {
occupancy = (Integer) resp.get("occupancy");
} catch (JSONException e) {
e.printStackTrace();
}
}
if (latch != null)
latch.countDown();
}
@Override
public void errorCallback(String channel, PubnubError error) {
if (latch != null)
latch.countDown();
}
}
class HistoryCallback extends Callback {
private int count;
private String[] uuids;
private CountDownLatch latch;
public int getCount() {
return count;
}
public HistoryCallback(CountDownLatch latch) {
this.latch = latch;
}
public HistoryCallback() {
}
@Override
public void successCallback(String channel, Object message) {
JSONArray resp = null;
try {
resp = (new JSONArray(message.toString())).getJSONArray(0);
} catch (JSONException e1) {
e1.printStackTrace();
}
if (resp != null) {
count = resp.length();
}
if (latch != null)
latch.countDown();
}
@Override
public void errorCallback(String channel, PubnubError error) {
if (latch != null)
latch.countDown();
}
}
Pubnub pubnub = new Pubnub("demo", "demo");
Pubnub pubnub_enc = new Pubnub("demo", "demo", "demo", "demo", false);
String testSuccessMessage = "";
@Test
public void testPublishString() throws PubnubException, InterruptedException {
String channel = "java-unittest-" + Math.random();
final String sendMessage = "Test Message " + Math.random();
final CountDownLatch latch = new CountDownLatch(2);
final PublishCallback pbCb = new PublishCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, sendMessage, pbCb);
}
};
pubnub.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, pbCb.getResult());
assertEquals(sendMessage, sbCb.getResponse());
}
@Test
public void testPublishJSONArray() throws PubnubException, InterruptedException {
String channel = "java-unittest-" + Math.random();
final JSONArray sendMessage = new JSONArray().put(1).put("Test");
final CountDownLatch latch = new CountDownLatch(2);
final PublishCallback pbCb = new PublishCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, sendMessage, pbCb);
}
};
pubnub.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, pbCb.getResult());
assertEquals(sendMessage.toString(), sbCb.getResponse().toString());
}
@Test
public void testPublishJSONObject() throws InterruptedException, PubnubException, JSONException {
String channel = "java-unittest-" + Math.random();
final JSONObject sendMessage;
sendMessage = new JSONObject().put("1", "Test");
final CountDownLatch latch = new CountDownLatch(2);
final PublishCallback pbCb = new PublishCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, sendMessage, pbCb);
}
};
pubnub.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, pbCb.getResult());
assertEquals(sendMessage.toString(), sbCb.getResponse().toString());
}
@Test
public void testPublishStringWithEncryption() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final String sendMessage = "Test Message " + Math.random();
final CountDownLatch latch = new CountDownLatch(2);
final PublishCallback pbCb = new PublishCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
pubnub_enc.publish(channel, sendMessage, pbCb);
}
};
pubnub_enc.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, pbCb.getResult());
assertEquals(sendMessage, sbCb.getResponse());
}
@Test
public void testPublishJSONArrayWithEncryption() throws PubnubException, InterruptedException {
String channel = "java-unittest-" + Math.random();
final JSONArray sendMessage = new JSONArray().put(1).put("Test");
final CountDownLatch latch = new CountDownLatch(2);
final PublishCallback pbCb = new PublishCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
pubnub_enc.publish(channel, sendMessage, pbCb);
}
};
pubnub_enc.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, pbCb.getResult());
assertEquals(sendMessage.toString(), sbCb.getResponse().toString());
}
@Test
public void testPublishJSONObjectWithEncryption() throws InterruptedException, PubnubException, JSONException {
String channel = "java-unittest-" + Math.random();
final JSONObject sendMessage;
sendMessage = new JSONObject().put("1", "Test");
final CountDownLatch latch = new CountDownLatch(2);
final PublishCallback pbCb = new PublishCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
pubnub_enc.publish(channel, sendMessage, pbCb);
}
};
pubnub_enc.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, pbCb.getResult());
assertEquals(sendMessage.toString(), sbCb.getResponse().toString());
}
@Test
public void testHereNowOneUser() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(2);
final HereNowCallback hnCb = new HereNowCallback(latch);
SubscribeCallback sbCb = new SubscribeCallback(latch) {
@Override
public void connectCallback(String channel, Object message) {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
}
pubnub.hereNow(channel, hnCb);
pubnub.unsubscribe(channel);
if (latch != null)
latch.countDown();
}
};
pubnub.subscribe(channel, sbCb);
latch.await(10, TimeUnit.SECONDS);
assertEquals(1, hnCb.getOccupancy());
}
@Test
public void testHistoryCountOne() {
String channel = "java-unittest-" + Math.random();
final String sendMessage = "Test Message " + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final int limit = 1;
final HistoryCallback hCb = new HistoryCallback(latch);
pubnub.publish(channel, sendMessage, new PublishCallback(latch) {
@Override
public void successCallback(String channel, Object message) {
pubnub.publish(channel, sendMessage,
new PublishCallback(latch) {
@Override
public void successCallback(String channel,
Object message) {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
pubnub.history(channel, limit, hCb);
super.successCallback(channel, message);
}
});
super.successCallback(channel, message);
}
});
try {
latch.await(10, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
assertEquals(limit, hCb.getCount());
}
@Test
public void testHistory() {
String channel = "java-unittest-" + Math.random();
final String sendMessage = "Test Message " + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final HistoryCallback hCb = new HistoryCallback(latch);
pubnub.publish(channel, sendMessage, new PublishCallback(latch) {
@Override
public void successCallback(String channel, Object message) {
pubnub.publish(channel, sendMessage,
new PublishCallback(latch) {
@Override
public void successCallback(String channel,
Object message) {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
pubnub.history(channel, 100, hCb);
super.successCallback(channel, message);
}
});
super.successCallback(channel, message);
}
});
try {
latch.await(10, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
assertEquals(2, hCb.getCount());
}
@Test
public void testConnectionRestore() throws PubnubException, InterruptedException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(true);
pubnub.subscribe(channel, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertEquals(10, subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreUnsubInSuccessCallback() throws PubnubException, InterruptedException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(5);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(true);
pubnub.subscribe(channel, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 20, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertEquals(20, subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreMultipleChannels() throws PubnubException, InterruptedException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(true);
pubnub.subscribe(channel, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(5000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertEquals(10, subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreMultipleChannelsUnsubInSuccessCallback() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(5);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(true);
pubnub.subscribe(channel, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 20, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(5000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertEquals(20, subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreFalse() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(2);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(false);
pubnub.subscribe(channel, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertNull(subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreFalseMultipleChannels() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(false);
pubnub.subscribe(new String[]{channel, "b"}, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertNull(subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreFalseUnsubInSuccessCallback() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(false);
pubnub.subscribe(channel, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 20, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertNull(subscribeCb.getResponse());
}
@Test
public void testConnectionRestoreFalseMultipleChannelsUnsubInSuccessCallback() throws InterruptedException, PubnubException {
String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(3);
final SubscribeCallback subscribeCb = new SubscribeCallback(latch);
pubnub.setResumeOnReconnect(false);
pubnub.subscribe(new String[]{channel, "b"}, new SubscribeCallback(latch) {
public void connectCallback(String channel, Object message) {
pubnub.publish(channel, 10, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
pubnub.unsubscribe(channel);
pubnub.publish(channel, 20, new PublishCallback(latch) {
public void successCallback(String channel, Object message) {
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
pubnub.subscribe(channel, subscribeCb);
} catch (PubnubException e) {
e.printStackTrace();
}
latch.countDown();
}
});
latch.countDown();
}
});
latch.countDown();
}
});
latch.await(10, TimeUnit.SECONDS);
assertNull(subscribeCb.getResponse());
}
@Test
public void testSubscribeInMultipleThreads() throws InterruptedException {
final String channel = "java-unittest-" + Math.random();
final CountDownLatch latch = new CountDownLatch(100);
final Hashtable