git.iko.soy/WebSocketTest
on master help sign in
swap
b6b2e105 ahead, 0 behind
13 files changed+375−2
since splitsnapshots
unifiedsplit
Procfile+1 Sources/App/Models/CyclicBuffer.swift+40 Sources/App/Models/EventResponse.swift+30 Sources/App/Models/Events.swift+37 Sources/App/Models/Post.swift+23 Sources/App/Models/PostEvent.swift+37 Sources/App/Models/SafeDict.swift+59 Sources/App/Models/User.swift+28 Sources/App/boot.swift+59−1 Sources/App/configure.swift+18 Sources/App/routes.swift+36 Sources/Run/main.swift+6 cloud.yml+1−1
Procfilenew file+1
@@ -0,0 +1 @@
1+web: Run serve --env production --port $PORT --hostname 0.0.0.0
Sources/App/Models/CyclicBuffer.swiftnew file+40
@@ -0,0 +1,40 @@
1+//
2+// CyclicBuffer.swift
3+// App
4+//
5+// Created by Ilya on 29/11/2018.
6+//
7+
8+import Foundation
9+
10+struct CyclicBuffer<T> {
11+ private var buffer: [T?]
12+ private var index = 0
13+ let size: Int
14+
15+ init(count: Int) {
16+ buffer = Array(repeating: nil, count: count)
17+ size = count
18+ }
19+
20+ private let queue = DispatchQueue(label: "soy.iko.cyclicBufferqueue", attributes: DispatchQueue.Attributes.concurrent)
21+
22+ mutating func append(_ element: T) {
23+ queue.sync(flags: .barrier) {
24+ self.buffer[self.index] = element
25+ self.index = (self.index + 1) % self.size
26+ }
27+ }
28+
29+ func getLast(count: Int = 1) -> [T] {
30+ var out: [T] = []
31+
32+ queue.sync {
33+ out = Array((buffer[index..<size] + buffer[0..<index]).lazy
34+ .compactMap {$0}
35+ .suffix(count))
36+ }
37+
38+ return out
39+ }
40+}
Sources/App/Models/EventResponse.swiftnew file+30
@@ -0,0 +1,30 @@
1+//
2+// EventResponse.swift
3+// App
4+//
5+// Created by Ilya on 29/11/2018.
6+//
7+
8+import Foundation
9+import Vapor
10+
11+struct EventResponse: Encodable, Content {
12+
13+ let eventToken: Events<PostEvent>.Token
14+ let updates: [PostEvent]?
15+
16+ init(token: Events<PostEvent>.Token?) {
17+ if let token = token,
18+ PostEvent.events.isValid(token: token) {
19+ updates = PostEvent.events.getEvents(since: token)
20+ } else {
21+ updates = nil
22+ }
23+ eventToken = PostEvent.events.lastId
24+ }
25+
26+ init(update: PostEvent, token: Events<PostEvent>.Token) {
27+ self.eventToken = token
28+ self.updates = [update]
29+ }
30+}
Sources/App/Models/Events.swiftnew file+37
@@ -0,0 +1,37 @@
1+//
2+// Model.swift
3+// App
4+//
5+// Created by Ilya on 29/11/2018.
6+//
7+
8+import Foundation
9+
10+struct Events<T> {
11+ private var buffer: CyclicBuffer<T>
12+
13+ init(count: Int) {
14+ buffer = CyclicBuffer(count: count)
15+ }
16+
17+ private(set) var lastId: Token = 0
18+
19+ typealias Token = Int
20+
21+ mutating func append(_ event: T) -> Token {
22+ lastId += 1
23+
24+ buffer.append(event)
25+
26+ return lastId
27+ }
28+
29+ func getEvents(since id: Token) -> [T] {
30+ return buffer.getLast(count: lastId - id)
31+ }
32+
33+ func isValid(token: Token) -> Bool {
34+ let dif = (lastId - token)
35+ return (dif < buffer.size) && (dif > 0)
36+ }
37+}
Sources/App/Models/Post.swiftnew file+23
@@ -0,0 +1,23 @@
1+//
2+// Post.swift
3+// App
4+//
5+// Created by Ilya on 29/11/2018.
6+//
7+
8+import Foundation
9+import Vapor
10+
11+struct Post: Encodable, Content {
12+ typealias Id = Int
13+ let text: String
14+ let user: User.Id
15+ let id: Id = {
16+ lastId += 1
17+ return lastId
18+ }()
19+
20+ private static var lastId = 0
21+
22+ static var storage = SafeDict<Id, Post>()
23+}
Sources/App/Models/PostEvent.swiftnew file+37
@@ -0,0 +1,37 @@
1+//
2+// PostEvent.swift
3+// App
4+//
5+// Created by Ilya on 29/11/2018.
6+//
7+
8+import Foundation
9+import Vapor
10+
11+enum PostEvent: Encodable, Content {
12+ init(from decoder: Decoder) throws {
13+ fatalError()
14+ }
15+
16+ func encode(to encoder: Encoder) throws {
17+ var container = encoder.container(keyedBy: CodingKeys.self)
18+ switch self {
19+ case .added(let id):
20+ try container.encode(id, forKey: .added)
21+ case .deleted(let id):
22+ try container.encode(id, forKey: .deleted)
23+ }
24+ }
25+
26+ private enum CodingKeys: String, CodingKey {
27+ case added
28+ case deleted
29+ }
30+
31+ public static var events = Events<PostEvent>(count: 1000)
32+
33+ case added(Post.Id)
34+ case deleted(Post.Id)
35+}
36+
37+
Sources/App/Models/SafeDict.swiftnew file+59
@@ -0,0 +1,59 @@
1+//
2+// File.swift
3+// App
4+//
5+// Created by Ilya on 30/11/2018.
6+//
7+
8+import Foundation
9+
10+func getRandomNum(_ min: Int, _ max: Int) -> Int {
11+ #if os(Linux)
12+ return Int(random() % max) + min
13+ #else
14+ return Int(arc4random_uniform(UInt32(max)) + UInt32(min))
15+ #endif
16+}
17+
18+struct SafeDict<Key: Hashable, Value> {
19+ private var queue = DispatchQueue(label: "soy.iko.safeDict.queue", attributes: .concurrent)
20+
21+ private var dict = Dictionary<Key, Value>()
22+
23+ subscript(_ key: Key) -> Value? {
24+ get {
25+ var out: Value?
26+
27+ queue.sync {
28+ out = dict[key]
29+ }
30+
31+ return out
32+ }
33+
34+ set {
35+ queue.sync(flags: .barrier) {
36+ dict[key] = newValue
37+ }
38+ }
39+ }
40+
41+ mutating func removeRandom() -> Key? {
42+ var key: Key?
43+ queue.sync(flags: .barrier) {
44+ if dict.count > 0 {
45+ key = Array(dict.keys)[getRandomNum(0, dict.count)]
46+ dict[key!] = nil
47+ }
48+ }
49+ return key
50+ }
51+
52+ var count: Int {
53+ return dict.count
54+ }
55+
56+ var values: Dictionary<Key, Value>.Values {
57+ return dict.values
58+ }
59+}
Sources/App/Models/User.swiftnew file+28
@@ -0,0 +1,28 @@
1+//
2+// User.swift
3+// App
4+//
5+// Created by Ilya on 29/11/2018.
6+//
7+
8+import Foundation
9+import Vapor
10+
11+struct User: Encodable, Content {
12+ struct Name: Content {
13+ let first: String
14+ let second: String
15+ }
16+
17+ typealias Id = Int
18+
19+ let name: Name
20+ let id: Id = {
21+ lastID += 1
22+ return lastID
23+ }()
24+
25+ static var storage = SafeDict<Id, User>()
26+
27+ private static var lastID = 0
28+}
Sources/App/boot.swift+59−1
@@ -1,6 +1,64 @@
11import Vapor
22
3+struct OpenSockets {
4+ private(set) static var sockets = SafeDict<Int, WebSocket>()
5+
6+ static func insert(_ socket: WebSocket) -> Int {
7+ let out = id
8+ id += 1
9+ sockets[out] = socket
10+ return out
11+ }
12+
13+ static func remove(at id: Int) {
14+ sockets[id] = nil
15+ }
16+
17+
18+ private static var id = 0
19+}
20+
21+
22+let enc: JSONEncoder = {
23+ var out = JSONEncoder()
24+ return out
25+}()
26+
327/// Called after your application has initialized.
428public func boot(_ app: Application) throws {
5− // your code here
29+ let akakiy = User(name: User.Name(first: "Акакий", second: "Программистович"))
30+ User.storage[akakiy.id] = akakiy
31+
32+ app.eventLoop.scheduleRepeatedTask(initialDelay: .seconds(10), delay: .seconds(10)) { (task: RepeatedTask) in
33+ let _ = try app.client().get("https://baconipsum.com/api/?type=all-meat&paras=1&start-with-lorem=0")
34+ .flatMap { try $0.content.decode([String].self) }
35+ .do { (texts) in
36+ guard let text = texts.first else { return }
37+
38+ let p = Post.init(text: text, user: akakiy.id)
39+
40+ Post.storage[p.id] = p
41+
42+ let event = PostEvent.added(p.id)
43+ let t = PostEvent.events.append(event)
44+
45+ OpenSockets.sockets.values.forEach { (ws: WebSocket) in
46+ ws.send(try! enc.encode(EventResponse(update: event, token: t)))
47+ }
48+ }
49+ }
50+
51+ let _ = app.eventLoop.scheduleRepeatedTask(initialDelay: .seconds(205), delay: .seconds(10)) { (task: RepeatedTask) throws -> Void in
52+ if (Post.storage.count > 20) {
53+ if let key = Post.storage.removeRandom() {
54+ let event = PostEvent.deleted(key)
55+ let t = PostEvent.events.append(event)
56+ OpenSockets.sockets.values.forEach { (ws: WebSocket) in
57+ ws.send(try! enc.encode(EventResponse(update: event, token: t)))
58+ }
59+ }
60+ }
61+
62+ return
63+ }
664}
Sources/App/configure.swift+18
@@ -38,5 +38,23 @@ public func configure(_ config: inout Config, _ env: inout Environment, _ servic
3838 })
3939 }
4040
41+ ws.get("postUpdates", Int.parameter) { (ws, req) in
42+ let i = OpenSockets.insert(ws)
43+ ws.onClose.always {
44+ OpenSockets.remove(at: i)
45+ }
46+ let update = EventResponse(token: try! req.parameters.next(Int.self))
47+ ws.send(try! enc.encode(update))
48+ }
49+
50+ ws.get("postUpdates") { (ws, req) in
51+ let i = OpenSockets.insert(ws)
52+ ws.onClose.always {
53+ OpenSockets.remove(at: i)
54+ }
55+ let update = EventResponse(token: nil)
56+ ws.send(try! enc.encode(update))
57+ }
58+
4159 services.register(ws, as: WebSocketServer.self)
4260}
Sources/App/routes.swift+36
@@ -17,4 +17,40 @@ public func routes(_ router: Router) throws {
1717 router.get("todos", use: todoController.index)
1818 router.post("todos", use: todoController.create)
1919 router.delete("todos", Todo.parameter, use: todoController.delete)
20+
21+ router.get("user", Int.parameter) { (req) -> User in
22+ let id = try req.parameters.next(Int.self)
23+ guard let user = User.storage[id] else {
24+ throw DatabaseError.notFound
25+ }
26+ return user
27+ }
28+
29+ router.get("post", Int.parameter) { req -> Post in
30+ let id = try req.parameters.next(Int.self)
31+ guard let post = Post.storage[id] else {
32+ throw DatabaseError.notFound
33+ }
34+ return post
35+ }
36+
37+ router.get("posts") { (req) -> [Post] in
38+ return Array(Post.storage.values)
39+ }
40+}
41+
42+enum DatabaseError: Error, AbortError {
43+ var status: HTTPResponseStatus {
44+ return .notFound
45+ }
46+
47+ var reason: String {
48+ return "Requested entry not found."
49+ }
50+
51+ var identifier: String {
52+ return "Requested entry not found."
53+ }
54+
55+ case notFound
2056}
Sources/Run/main.swift+6
@@ -1,3 +1,9 @@
11import App
22
3+#if os(Linux)
4+import SwiftGlibc
5+srandom(UInt32(time(nil)))
6+#endif
7+
8+
39try app(.detect()).run()
cloud.yml+1−1
@@ -1,3 +1,3 @@
11type: "vapor"
2−swift_version: "4.1.0"
2+swift_version: "4.2.0"
33run_parameters: "serve --port 8080 --hostname 0.0.0.0"