Skip to content
This repository was archived by the owner on Oct 5, 2023. It is now read-only.

Commit e493ef7

Browse files
SlashNephySlashNephy
authored
Merge pull request #21 from SlashNephy/dev
タイムシフトコメント再生時のシークを改善
2 parents 7f82fb3 + 1289c62 commit e493ef7

18 files changed

+669
-304
lines changed

‎docs/ws.html

+93
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
<html lang="en">
2+
<head>
3+
<title>saya WebSockets Test</title>
4+
<style>
5+
body {
6+
background-color: #708090;
7+
}
8+
</style>
9+
</head>
10+
<body>
11+
<div>
12+
<label for="url">URL: </label><input type="text" id="url" size="100" value="ws://localhost:1017/comments/jk1/live">
13+
<button type="button" onclick="connect();">Connect</button>
14+
<button type="button" onclick="disconnect(true);">Disconnect</button>
15+
</div>
16+
17+
<div>
18+
<label for="command">Command: </label><input type="text" id="command" size="90" value="{&quot;action&quot;: &quot;Sync&quot;, &quot;seconds&quot;: 0.0}">
19+
<button type="button" onclick="send();">Send</button>
20+
</div>
21+
22+
<div>Status: <span id="status"></span></div>
23+
24+
<hr>
25+
26+
<code id="response"></code>
27+
28+
<script>
29+
const url = document.getElementById("url");
30+
const command = document.getElementById("command");
31+
const status = document.getElementById("status");
32+
const response = document.getElementById("response");
33+
let retry = true;
34+
35+
let ws;
36+
function connect() {
37+
disconnect(false);
38+
39+
ws = new WebSocket(url.value);
40+
localStorage.setItem("url", url.value);
41+
42+
ws.onopen = () => {
43+
status.textContent = "OPEN";
44+
};
45+
ws.onclose = () => {
46+
status.textContent = "CLOSE";
47+
if (retry) {
48+
setTimeout(connect, 3000);
49+
}
50+
};
51+
ws.onerror = (e) => {
52+
status.textContent = `ERROR: ${e}`;
53+
};
54+
ws.onmessage = (e) => {
55+
const ele = document.createElement("p");
56+
ele.textContent = e.data;
57+
response.prepend(ele);
58+
};
59+
}
60+
61+
function disconnect(byUser) {
62+
try {
63+
ws.close();
64+
} catch {
65+
}
66+
67+
response.textContent = "";
68+
if (byUser) {
69+
retry = false;
70+
}
71+
}
72+
73+
function send() {
74+
try {
75+
ws.send(command.value);
76+
localStorage.setItem("command", command.value);
77+
} catch {
78+
}
79+
}
80+
81+
const lastUrl = localStorage.getItem("url");
82+
if (lastUrl) {
83+
url.value = lastUrl;
84+
}
85+
const lastCommand = localStorage.getItem("command");
86+
if (lastCommand) {
87+
command.value = lastCommand;
88+
}
89+
90+
connect();
91+
</script>
92+
</body>
93+
</html>

‎src/main/kotlin/blue/starry/saya/endpoints/Comments.kt

+9-133
Original file line numberDiff line numberDiff line change
@@ -3,34 +3,25 @@ package blue.starry.saya.endpoints
33
import blue.starry.saya.common.createSayaLogger
44
import blue.starry.saya.common.rejectWs
55
import blue.starry.saya.common.respondOr404
6-
import blue.starry.saya.models.Comment
76
import blue.starry.saya.models.CommentSource
87
import blue.starry.saya.models.TimeshiftCommentControl
9-
import blue.starry.saya.services.CommentChannelManager
10-
import blue.starry.saya.services.SayaMiyouTVApi
11-
import blue.starry.saya.services.miyoutv.toSayaComment
8+
import blue.starry.saya.services.comments.CommentChannelManager
129
import blue.starry.saya.services.nicojk.NicoJkApi
13-
import blue.starry.saya.services.nicolive.toSayaComment
1410
import io.ktor.application.*
1511
import io.ktor.http.*
1612
import io.ktor.http.cio.websocket.*
1713
import io.ktor.response.*
1814
import io.ktor.routing.*
1915
import io.ktor.util.*
2016
import io.ktor.websocket.*
21-
import kotlinx.coroutines.channels.Channel
2217
import kotlinx.coroutines.channels.consumeEach
23-
import kotlinx.coroutines.delay
24-
import kotlinx.coroutines.flow.collect
2518
import kotlinx.coroutines.flow.consumeAsFlow
2619
import kotlinx.coroutines.flow.filterIsInstance
27-
import kotlinx.coroutines.launch
20+
import kotlinx.coroutines.flow.mapNotNull
2821
import kotlinx.serialization.decodeFromString
2922
import kotlinx.serialization.encodeToString
3023
import kotlinx.serialization.json.Json
3124
import mu.KotlinLogging
32-
import java.util.concurrent.atomic.AtomicLong
33-
import kotlin.math.roundToLong
3425

3526
private val logger = KotlinLogging.createSayaLogger("saya.endpoints")
3627
private val jsonWithDefault = Json {
@@ -45,7 +36,6 @@ fun Route.wsLiveCommentsByTarget() {
4536
val channel = CommentChannelManager.findByTarget(target) ?: return@webSocket rejectWs { "Parameter target is invalid." }
4637

4738
CommentChannelManager.subscribeLiveComments(channel, sources).consumeEach {
48-
4939
send(jsonWithDefault.encodeToString(it))
5040
}
5141
}
@@ -54,136 +44,22 @@ fun Route.wsLiveCommentsByTarget() {
5444
fun Route.wsTimeshiftCommentsByTarget() {
5545
webSocket {
5646
val target: String by call.parameters
57-
58-
// エポック秒
5947
val startAt: Long by call.parameters
6048
val endAt: Long by call.parameters
61-
val duration = (endAt - startAt).toInt()
62-
6349
val sources = CommentSource.from(call.parameters["sources"])
6450

6551
val channel = CommentChannelManager.findByTarget(target) ?: return@webSocket rejectWs { "Parameter target is invalid." }
66-
val timeMs = AtomicLong(startAt * 1000)
67-
var pause = true
68-
69-
// unit 秒ずつ分割して取得
70-
val unit = 600
71-
72-
// コメント配信ループ
73-
suspend fun Channel<Comment>.sendLoop() {
74-
consumeEach { comment ->
75-
val currentMs = comment.time * 1000 + comment.timeMs
76-
val waitMs = currentMs - timeMs.get()
77-
if (waitMs < -10000) {
78-
logger.trace { "破棄 ($waitMs) : $comment" }
79-
return@consumeEach
80-
}
81-
82-
delay(waitMs)
83-
84-
while (pause) {
85-
delay(100)
86-
}
87-
send(jsonWithDefault.encodeToString(comment))
88-
logger.trace { "配信: $comment" }
89-
90-
timeMs.getAndUpdate { prev ->
91-
maxOf(prev, currentMs)
92-
}
93-
}
94-
}
95-
96-
if (CommentSource.Gochan in sources && channel.miyoutvId != null) {
97-
launch {
98-
val client = SayaMiyouTVApi ?: return@launch
99-
val queue = Channel<Comment>(Channel.UNLIMITED)
100-
101-
launch {
102-
repeat(duration / unit) { i ->
103-
client.getComments(
104-
channel.miyoutvId,
105-
(startAt + unit * i) * 1000,
106-
minOf(startAt + unit * (i + 1), endAt) * 1000
107-
).data.comments.map {
108-
it.toSayaComment()
109-
}.forEach {
110-
queue.send(it)
111-
}
112-
}
113-
}
114-
115-
queue.sendLoop()
116-
}
117-
}
118-
119-
if (CommentSource.Nicolive in sources && channel.nicojkId != null) {
120-
launch {
121-
val queue = Channel<Comment>(Channel.UNLIMITED)
122-
123-
launch {
124-
repeat(duration / unit) { i ->
125-
NicoJkApi.getComments(
126-
"jk${channel.nicojkId}",
127-
startAt + unit * i,
128-
minOf(startAt + unit * (i + 1), endAt)
129-
).packets.map {
130-
it.chat.toSayaComment(
131-
source = "ニコニコ実況過去ログAPI [jk${channel.nicojkId}]",
132-
sourceUrl = "https://jikkyo.tsukumijima.net/api/kakolog/jk${channel.nicojkId}?starttime=${it.chat.date}&endtime=${it.chat.date + 1}&format=json"
133-
)
134-
}.forEach {
135-
queue.send(it)
136-
}
137-
}
138-
}
139-
140-
queue.sendLoop()
141-
}
142-
}
143-
144-
// WS コントロール処理ループ
145-
incoming.consumeAsFlow().filterIsInstance<Frame.Text>().collect {
146-
val control = try {
52+
val controls = incoming.consumeAsFlow().filterIsInstance<Frame.Text>().mapNotNull {
53+
try {
14754
Json.decodeFromString<TimeshiftCommentControl>(it.readText())
14855
} catch (t: Throwable) {
149-
logger.error(t) { "WS コントロール命令のパースに失敗しました。" }
150-
return@collect
151-
}
152-
153-
when (control.action) {
154-
/**
155-
* クライアントの準備ができ, コメントの配信を開始する命令
156-
* {"action": "Ready"}
157-
*
158-
* コメントの配信を再開する命令
159-
* {"action": "Resume"}
160-
*/
161-
TimeshiftCommentControl.Action.Ready,
162-
TimeshiftCommentControl.Action.Resume -> {
163-
pause = false
164-
}
165-
166-
/**
167-
* コメントの配信を一時停止する命令
168-
* {"action": "Pause"}
169-
*/
170-
//
171-
TimeshiftCommentControl.Action.Pause -> {
172-
pause = true
173-
}
174-
175-
/**
176-
* コメントの位置を同期する命令
177-
* {"action": "Sync", "seconds": 10.0}
178-
*/
179-
TimeshiftCommentControl.Action.Sync -> {
180-
pause = false
181-
182-
timeMs.set(((startAt + control.seconds) * 1000).roundToLong())
183-
}
56+
logger.error(t) { "TimeshiftCommentControl のパースに失敗しました: $it" }
57+
null
18458
}
59+
}
18560

186-
logger.debug { "クライアントの命令: $control" }
61+
CommentChannelManager.subscribeTimeshiftComments(channel, sources, controls, startAt, endAt).consumeEach {
62+
send(jsonWithDefault.encodeToString(it))
18763
}
18864
}
18965
}

‎src/main/kotlin/blue/starry/saya/endpoints/Definitions.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package blue.starry.saya.endpoints
22

3-
import blue.starry.saya.services.CommentChannelManager
3+
import blue.starry.saya.services.comments.CommentChannelManager
44
import io.ktor.application.*
55
import io.ktor.response.*
66
import io.ktor.routing.*

‎src/main/kotlin/blue/starry/saya/models/Comments.kt

+5
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,11 @@ data class Comment(
2727
*/
2828
val timeMs: Int,
2929

30+
/**
31+
* タイムシフトコメントでの開始地点からの再生時間 (秒)
32+
*/
33+
val seconds: Double? = null,
34+
3035
/**
3136
* コメントの投稿者名 / ユーザ ID
3237
*/

0 commit comments

Comments
 (0)