Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

fixed some tests and some code

  • Loading branch information...
commit dadf7da93d4c6d75d7ffb316c410e8c007ed3626 1 parent ffa2623
@Raynos Raynos authored
View
6 .testem.json
@@ -3,7 +3,7 @@
"./**/*.js"
],
"before_tests": "npm run build",
- "on_exit": "rm test/bundle.js",
- "test_page": "test/index.html",
- "launch_in_dev": ["phantomjs"]
+ "on_exit": "rm test/static/bundle.js",
+ "test_page": "test/static/index.html",
+ "launch_in_dev": []
}
View
11 package.json
@@ -23,14 +23,17 @@
"end-stream": "~0.1.0",
"expiry-model": "~1.0.1",
"re-emitter": "~0.3.0",
- "xtend": "~1.0.3"
+ "xtend": "~1.0.3",
+ "split-every": "0.0.1",
+ "node-uuid": "~1.4.0",
+ "after": "~0.6.0"
},
"devDependencies": {
"browserify-server": "~2.1.12",
"pubnub": "~3.3.1",
- "testem": "~0.2.50",
+ "testem": "~0.2.54",
"browserify": "~1.17.1",
- "tape": "~0.1.5",
+ "tape": "Raynos/tape",
"write-stream": "~0.4.3",
"pubnub-browserify": "~3.3.0",
"node-uuid": "~1.4.0",
@@ -43,6 +46,6 @@
}
],
"scripts": {
- "build": "browserify test/index.js -o test/bundle.js"
+ "build": "browserify test/index.js --debug -o test/static/bundle.js"
}
}
View
64 publish.js
@@ -1,25 +1,81 @@
var EndStream = require("end-stream")
var Queue = require("push-queue")
+var splitEvery = require("split-every")
+var uuid = require("node-uuid")
+var after = require("after")
module.exports = publish
function publish(client, channel) {
var push = Queue(sendItem)
- var stream = EndStream(push)
+ var stream = EndStream(function write(item, callback) {
+ push(JSON.stringify(item), callback)
+ })
return stream
- function sendItem(item, callback) {
+ function sendItem(json, callback) {
+ if (json.length > 600) {
+ var list = json.split("")
+ var id = uuid()
+ var chunks = splitEvery(550, list).
+ map(function (list) {
+ return list.join("")
+ })
+ var length = chunks.length
+ var enqueue = Queue(function (json, callback) {
+ publish(json, function (err) {
+ if (err) {
+ return stream.emit("error", err)
+ }
+
+ callback()
+ })
+ })
+
+ var done = after(length, function (err) {
+ if (err) {
+ return stream.emit("error", err)
+ }
+
+ callback()
+ })
+
+ chunks.forEach(function (chunk, index) {
+ var head = id + ":"
+
+ if (index === 0) {
+ head += length
+ }
+
+ head += "\n"
+
+ var data = head + chunk
+
+ enqueue(data, done)
+ })
+ } else {
+ publish("\n" + json, function (err) {
+ if (err) {
+ return stream.emit("error", err)
+ }
+
+ callback()
+ })
+ }
+ }
+
+ function publish(payload, callback) {
client.publish({
channel: channel
- , message: item
+ , message: payload
, callback: handleError
})
function handleError(info) {
if (info[0] === 0) {
var message = info[1] || "pubsub publish failed"
- stream.emit("error", Error(message))
+ return callback(Error(message))
}
callback()
View
30 subscribe.js
@@ -9,6 +9,7 @@ module.exports = subscribe
function subscribe(client, channel, options) {
var queue = ReadStream()
var stream = queue.stream
+ var hash = {}
options = options || {}
@@ -30,7 +31,34 @@ function subscribe(client, channel, options) {
options.reconnect && options.reconnect()
}
- , callback: queue.push
+ , callback: function (msg) {
+ var parts = msg.split("\n")
+ var header = parts[0]
+ var payload = parts[1]
+
+ if (header.length === 0) {
+ return queue.push(JSON.parse(payload))
+ }
+
+ var headerParts = header.split(":")
+ var id = headerParts[0]
+ var list = hash[id] || (hash[id] = [])
+ var length = +headerParts[1] || list[0]
+
+ if (list.length === 0) {
+ list[0] = length
+ }
+
+ list.push(payload)
+
+ if (list.length !== length + 1) {
+ return
+ }
+
+ var message = list.slice(1).join("")
+
+ queue.push(JSON.parse(message))
+ }
, error: function (info) {
var message = info[1] || "pubsub subscribe failed"
stream.emit("error", Error(message))
View
6 test/subscribe.js → test/duplex.js
@@ -2,12 +2,12 @@ var test = require("tape")
var uuid = require("node-uuid")
var after = require("after")
-var connect = require("./connect")
-var throughput = require("./throughput")
+var connect = require("./util/connect")
+var throughput = require("./util/throughput")
var duplex = require("../duplex")
-var channel = "pubnubstreamtestsubscribe" + uuid()
+var channel = "pubnub stream test duplex" + uuid()
test("can subscribe to channel", function (assert) {
var client = connect()
View
1  test/history.js
@@ -9,6 +9,7 @@ var history = require("../history")
var channel = "pubnubstreamtesthistory" + uuid()
+// not tested yet
test("can read from history", function (assert) {
var client = connect()
var stream = publish(client, channel)
View
3  test/index.js
@@ -1 +1,2 @@
-require("./subscribe")
+require("./duplex")
+require("./publish")
View
99 test/publish.js
@@ -0,0 +1,99 @@
+var test = require("tape")
+var uuid = require("node-uuid")
+var WriteStream = require("write-stream")
+
+var connect = require("./util/connect")
+
+var subscribe = require("../subscribe")
+var publish = require("../publish")
+
+var channel = "pubnub stream test publish" + uuid()
+
+test("can publish to channel", function (assert) {
+ var client = connect()
+ var read = subscribe(client, channel)
+
+ read.once("connect", function () {
+ var write = publish(client, channel)
+
+ write.write("one")
+ write.end()
+ })
+
+ read.pipe(WriteStream(function write(message) {
+ assert.equal(message, "one")
+ read.close()
+
+ assert.end()
+ }))
+})
+
+test("can publish multiple messages", function (assert) {
+ var client = connect()
+ var read = subscribe(client, channel)
+ var messages = ["one", "two", "three"]
+ var list = []
+
+ read.once("connect", function () {
+ var write = publish(client, channel)
+
+ messages.forEach(function (msg) {
+ write.write(msg)
+ })
+ write.end()
+ })
+
+ read.pipe(WriteStream(function write(message) {
+ list.push(message)
+
+ if (list.length === messages.length) {
+ finish()
+ }
+ }))
+
+ function finish() {
+ assert.deepEqual(list, messages)
+
+ read.close()
+
+ assert.end()
+ }
+})
+
+test("can send really large message", function (assert) {
+ assert.ok(true, "forces testem to play nice")
+ var message = largeMessage(1800)
+ var client = connect()
+ var read = subscribe(client, channel)
+
+ read.on("connect", function () {
+ var write = publish(client, channel)
+
+ write.write(message)
+ write.end()
+ })
+
+ read.pipe(WriteStream(function write(chunk) {
+ assert.equal(message, chunk)
+
+ read.close()
+ assert.end()
+ }))
+})
+
+function largeMessage(n) {
+ var res = ""
+ var cur = 0
+
+ for (var i = 0; i < n; i++) {
+ res += cur
+
+ cur++
+
+ if (cur === 10) {
+ cur = 0
+ }
+ }
+
+ return res
+}
View
1  test/index.html → test/static/index.html
@@ -3,6 +3,7 @@
<head>
<title>TAPE Example</title>
<script src="/testem.js"></script>
+ <script src="tap_adapter.js"></script>
<script src="bundle.js"></script>
</head>
<body>
View
51 test/static/tap_adapter.js
@@ -0,0 +1,51 @@
+(function () {
+ var Testem = window.Testem
+ var regex = /^((?:not )?ok) (\d+) (.+)$/
+
+ Testem.useCustomAdapter(tapAdapter)
+
+ function tapAdapter(socket){
+ var results = {
+ failed: 0
+ , passed: 0
+ , total: 0
+ , tests: []
+ }
+
+ socket.emit('tests-start')
+
+ Testem.handleConsoleMessage = function(msg){
+ var m = msg.match(regex)
+ if (m) {
+ var passed = m[1] === 'ok'
+ var test = {
+ passed: passed ? 1 : 0,
+ failed: passed ? 0 : 1,
+ total: 1,
+ id: m[2],
+ name: m[3],
+ items: []
+ }
+
+ if (passed) {
+ results.passed++
+ } else {
+ console.error("failure", m)
+
+ results.failed++
+ }
+
+ results.total++
+
+ socket.emit('test-result', test)
+ results.tests.push(test)
+ } else if (msg === '# ok' || msg.match(/^# tests \d+/)){
+ socket.emit('all-test-results', results)
+ }
+
+ // return false if you want to prevent the console message from
+ // going to the console
+ // return false
+ }
+ }
+}())
View
0  test/connect.js → test/util/connect.js
File renamed without changes
View
0  test/throughput.js → test/util/throughput.js
File renamed without changes
Please sign in to comment.
Something went wrong with that request. Please try again.