Send multiple events together
This commit is contained in:
parent
e723cc6d98
commit
735fccf043
@ -8,6 +8,7 @@ import it.tdlight.reactiveapi.Event.ClientBoundEvent;
|
|||||||
import it.tdlight.reactiveapi.Event.Request;
|
import it.tdlight.reactiveapi.Event.Request;
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.time.Instant;
|
import java.time.Instant;
|
||||||
|
import java.util.List;
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import reactor.core.publisher.BufferOverflowStrategy;
|
import reactor.core.publisher.BufferOverflowStrategy;
|
||||||
import reactor.core.publisher.Flux;
|
import reactor.core.publisher.Flux;
|
||||||
@ -26,9 +27,9 @@ public class AtomixReactiveApiMultiClient implements ReactiveApiMultiClient, Aut
|
|||||||
this.eventService = api.getAtomix().getEventService();
|
this.eventService = api.getAtomix().getEventService();
|
||||||
|
|
||||||
clientBoundEvents = Flux
|
clientBoundEvents = Flux
|
||||||
.<ClientBoundEvent>push(sink -> {
|
.<List<ClientBoundEvent>>push(sink -> {
|
||||||
var subscriptionFuture = eventService.subscribe("session-client-bound-events",
|
var subscriptionFuture = eventService.subscribe("session-client-bound-events",
|
||||||
LiveAtomixReactiveApiClient::deserializeEvent,
|
LiveAtomixReactiveApiClient::deserializeEvents,
|
||||||
s -> {
|
s -> {
|
||||||
sink.next(s);
|
sink.next(s);
|
||||||
return CompletableFuture.completedFuture(null);
|
return CompletableFuture.completedFuture(null);
|
||||||
@ -38,6 +39,7 @@ public class AtomixReactiveApiMultiClient implements ReactiveApiMultiClient, Aut
|
|||||||
sink.onDispose(() -> subscriptionFuture.thenAccept(Subscription::close));
|
sink.onDispose(() -> subscriptionFuture.thenAccept(Subscription::close));
|
||||||
}, OverflowStrategy.ERROR)
|
}, OverflowStrategy.ERROR)
|
||||||
.onBackpressureBuffer(0xFFFF, BufferOverflowStrategy.ERROR)
|
.onBackpressureBuffer(0xFFFF, BufferOverflowStrategy.ERROR)
|
||||||
|
.flatMapIterable(list -> list)
|
||||||
.takeUntil(s -> closed)
|
.takeUntil(s -> closed)
|
||||||
.share();
|
.share();
|
||||||
}
|
}
|
||||||
|
@ -8,6 +8,7 @@ import it.tdlight.reactiveapi.Event.ClientBoundEvent;
|
|||||||
import it.tdlight.reactiveapi.Event.Request;
|
import it.tdlight.reactiveapi.Event.Request;
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.time.Instant;
|
import java.time.Instant;
|
||||||
|
import java.util.List;
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import java.util.concurrent.atomic.AtomicLong;
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
import reactor.core.Disposable;
|
import reactor.core.Disposable;
|
||||||
@ -38,9 +39,9 @@ public class DynamicAtomixReactiveApiClient implements ReactiveApiClient, AutoCl
|
|||||||
this.userId = userId;
|
this.userId = userId;
|
||||||
|
|
||||||
clientBoundEvents = Flux
|
clientBoundEvents = Flux
|
||||||
.<ClientBoundEvent>push(sink -> {
|
.<List<ClientBoundEvent>>push(sink -> {
|
||||||
var subscriptionFuture = eventService.subscribe("session-client-bound-events",
|
var subscriptionFuture = eventService.subscribe("session-client-bound-events",
|
||||||
LiveAtomixReactiveApiClient::deserializeEvent,
|
LiveAtomixReactiveApiClient::deserializeEvents,
|
||||||
s -> {
|
s -> {
|
||||||
sink.next(s);
|
sink.next(s);
|
||||||
return CompletableFuture.completedFuture(null);
|
return CompletableFuture.completedFuture(null);
|
||||||
@ -49,9 +50,10 @@ public class DynamicAtomixReactiveApiClient implements ReactiveApiClient, AutoCl
|
|||||||
);
|
);
|
||||||
sink.onDispose(() -> subscriptionFuture.thenAccept(Subscription::close));
|
sink.onDispose(() -> subscriptionFuture.thenAccept(Subscription::close));
|
||||||
}, OverflowStrategy.ERROR)
|
}, OverflowStrategy.ERROR)
|
||||||
|
.onBackpressureBuffer(0xFFFF, BufferOverflowStrategy.ERROR)
|
||||||
|
.flatMapIterable(list -> list)
|
||||||
.filter(e -> e.userId() == userId)
|
.filter(e -> e.userId() == userId)
|
||||||
.doOnNext(e -> liveId.set(e.liveId()))
|
.doOnNext(e -> liveId.set(e.liveId()))
|
||||||
.onBackpressureBuffer(0xFFFF, BufferOverflowStrategy.ERROR)
|
|
||||||
.share();
|
.share();
|
||||||
|
|
||||||
liveIdChange = this.clientBoundEvents()
|
liveIdChange = this.clientBoundEvents()
|
||||||
|
@ -19,6 +19,8 @@ import java.io.DataOutputStream;
|
|||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.time.Instant;
|
import java.time.Instant;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import org.apache.commons.lang3.SerializationException;
|
import org.apache.commons.lang3.SerializationException;
|
||||||
import reactor.core.publisher.BufferOverflowStrategy;
|
import reactor.core.publisher.BufferOverflowStrategy;
|
||||||
@ -39,15 +41,20 @@ public class LiveAtomixReactiveApiClient implements ReactiveApiClient {
|
|||||||
this.liveId = liveId;
|
this.liveId = liveId;
|
||||||
this.userId = userId;
|
this.userId = userId;
|
||||||
this.clientBoundEvents = Flux
|
this.clientBoundEvents = Flux
|
||||||
.<ClientBoundEvent>push(sink -> {
|
.<List<ClientBoundEvent>>push(sink -> {
|
||||||
var subscriptionFuture = eventService.subscribe("session-client-bound-events", LiveAtomixReactiveApiClient::deserializeEvent, s -> {
|
var subscriptionFuture = eventService.subscribe("session-client-bound-events",
|
||||||
sink.next(s);
|
LiveAtomixReactiveApiClient::deserializeEvents,
|
||||||
return CompletableFuture.completedFuture(null);
|
s -> {
|
||||||
}, (a) -> null);
|
sink.next(s);
|
||||||
|
return CompletableFuture.completedFuture(null);
|
||||||
|
},
|
||||||
|
(a) -> null
|
||||||
|
);
|
||||||
sink.onDispose(() -> subscriptionFuture.thenAccept(Subscription::close));
|
sink.onDispose(() -> subscriptionFuture.thenAccept(Subscription::close));
|
||||||
}, OverflowStrategy.ERROR)
|
}, OverflowStrategy.ERROR)
|
||||||
.filter(e -> e.userId() == userId && e.liveId() == liveId)
|
|
||||||
.onBackpressureBuffer(0xFFFF, BufferOverflowStrategy.ERROR)
|
.onBackpressureBuffer(0xFFFF, BufferOverflowStrategy.ERROR)
|
||||||
|
.flatMapIterable(list -> list)
|
||||||
|
.filter(e -> e.userId() == userId && e.liveId() == liveId)
|
||||||
.share();
|
.share();
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -103,20 +110,39 @@ public class LiveAtomixReactiveApiClient implements ReactiveApiClient {
|
|||||||
static ClientBoundEvent deserializeEvent(byte[] bytes) {
|
static ClientBoundEvent deserializeEvent(byte[] bytes) {
|
||||||
try (var byteArrayInputStream = new ByteArrayInputStream(bytes)) {
|
try (var byteArrayInputStream = new ByteArrayInputStream(bytes)) {
|
||||||
try (var is = new DataInputStream(byteArrayInputStream)) {
|
try (var is = new DataInputStream(byteArrayInputStream)) {
|
||||||
var liveId = is.readLong();
|
return deserializeEvent(is);
|
||||||
var userId = is.readLong();
|
|
||||||
return switch (is.readByte()) {
|
|
||||||
case 0x01 -> new OnUpdateData(liveId, userId, (TdApi.Update) TdApi.Deserializer.deserialize(is));
|
|
||||||
case 0x02 -> new OnUpdateError(liveId, userId, (TdApi.Error) TdApi.Deserializer.deserialize(is));
|
|
||||||
case 0x03 -> new OnUserLoginCodeRequested(liveId, userId, is.readLong());
|
|
||||||
case 0x04 -> new OnBotLoginCodeRequested(liveId, userId, is.readUTF());
|
|
||||||
case 0x05 -> new OnOtherDeviceLoginRequested(liveId, userId, is.readUTF());
|
|
||||||
case 0x06 -> new OnPasswordRequested(liveId, userId, is.readUTF(), is.readBoolean(), is.readUTF());
|
|
||||||
default -> throw new IllegalStateException("Unexpected value: " + is.readByte());
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
} catch (IOException ex) {
|
} catch (IOException ex) {
|
||||||
throw new SerializationException(ex);
|
throw new SerializationException(ex);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static List<ClientBoundEvent> deserializeEvents(byte[] bytes) {
|
||||||
|
try (var byteArrayInputStream = new ByteArrayInputStream(bytes)) {
|
||||||
|
try (var is = new DataInputStream(byteArrayInputStream)) {
|
||||||
|
var len = is.readInt();
|
||||||
|
var result = new ArrayList<ClientBoundEvent>(len);
|
||||||
|
for (int i = 0; i < len; i++) {
|
||||||
|
result.add(deserializeEvent(is));
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
} catch (IOException ex) {
|
||||||
|
throw new SerializationException(ex);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static ClientBoundEvent deserializeEvent(DataInputStream is) throws IOException {
|
||||||
|
var liveId = is.readLong();
|
||||||
|
var userId = is.readLong();
|
||||||
|
return switch (is.readByte()) {
|
||||||
|
case 0x01 -> new OnUpdateData(liveId, userId, (TdApi.Update) TdApi.Deserializer.deserialize(is));
|
||||||
|
case 0x02 -> new OnUpdateError(liveId, userId, (TdApi.Error) TdApi.Deserializer.deserialize(is));
|
||||||
|
case 0x03 -> new OnUserLoginCodeRequested(liveId, userId, is.readLong());
|
||||||
|
case 0x04 -> new OnBotLoginCodeRequested(liveId, userId, is.readUTF());
|
||||||
|
case 0x05 -> new OnOtherDeviceLoginRequested(liveId, userId, is.readUTF());
|
||||||
|
case 0x06 -> new OnPasswordRequested(liveId, userId, is.readUTF(), is.readBoolean(), is.readUTF());
|
||||||
|
default -> throw new IllegalStateException("Unexpected value: " + is.readByte());
|
||||||
|
};
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
@ -177,10 +177,13 @@ public abstract class ReactiveApiPublisher {
|
|||||||
.cast(ClientBoundResultingEvent.class)
|
.cast(ClientBoundResultingEvent.class)
|
||||||
.map(ClientBoundResultingEvent::event)
|
.map(ClientBoundResultingEvent::event)
|
||||||
|
|
||||||
|
.limitRate(1)
|
||||||
|
.bufferTimeout(64, Duration.ofMillis(10))
|
||||||
|
|
||||||
// Send events to the client
|
// Send events to the client
|
||||||
.subscribeOn(Schedulers.parallel())
|
.subscribeOn(Schedulers.parallel())
|
||||||
.subscribe(clientBoundEvent -> eventService.broadcast("session-client-bound-events",
|
.subscribe(clientBoundEvent -> eventService.broadcast("session-client-bound-events",
|
||||||
clientBoundEvent, ReactiveApiPublisher::serializeEvent));
|
clientBoundEvent, ReactiveApiPublisher::serializeEvents));
|
||||||
|
|
||||||
publishedResultingEvents
|
publishedResultingEvents
|
||||||
// Obtain only cluster-bound events
|
// Obtain only cluster-bound events
|
||||||
@ -352,33 +355,12 @@ public abstract class ReactiveApiPublisher {
|
|||||||
return List.of();
|
return List.of();
|
||||||
}
|
}
|
||||||
|
|
||||||
private static byte[] serializeEvent(ClientBoundEvent clientBoundEvent) {
|
private static byte[] serializeEvents(List<ClientBoundEvent> clientBoundEvents) {
|
||||||
try (var byteArrayOutputStream = new ByteArrayOutputStream()) {
|
try (var byteArrayOutputStream = new ByteArrayOutputStream()) {
|
||||||
try (var dataOutputStream = new DataOutputStream(byteArrayOutputStream)) {
|
try (var dataOutputStream = new DataOutputStream(byteArrayOutputStream)) {
|
||||||
dataOutputStream.writeLong(clientBoundEvent.liveId());
|
dataOutputStream.writeInt(clientBoundEvents.size());
|
||||||
dataOutputStream.writeLong(clientBoundEvent.userId());
|
for (ClientBoundEvent clientBoundEvent : clientBoundEvents) {
|
||||||
if (clientBoundEvent instanceof OnUpdateData onUpdateData) {
|
writeClientBoundEvent(clientBoundEvent, dataOutputStream);
|
||||||
dataOutputStream.writeByte(0x1);
|
|
||||||
onUpdateData.update().serialize(dataOutputStream);
|
|
||||||
} else if (clientBoundEvent instanceof OnUpdateError onUpdateError) {
|
|
||||||
dataOutputStream.writeByte(0x2);
|
|
||||||
onUpdateError.error().serialize(dataOutputStream);
|
|
||||||
} else if (clientBoundEvent instanceof OnUserLoginCodeRequested onUserLoginCodeRequested) {
|
|
||||||
dataOutputStream.writeByte(0x3);
|
|
||||||
dataOutputStream.writeLong(onUserLoginCodeRequested.phoneNumber());
|
|
||||||
} else if (clientBoundEvent instanceof OnBotLoginCodeRequested onBotLoginCodeRequested) {
|
|
||||||
dataOutputStream.writeByte(0x4);
|
|
||||||
dataOutputStream.writeUTF(onBotLoginCodeRequested.token());
|
|
||||||
} else if (clientBoundEvent instanceof OnOtherDeviceLoginRequested onOtherDeviceLoginRequested) {
|
|
||||||
dataOutputStream.writeByte(0x5);
|
|
||||||
dataOutputStream.writeUTF(onOtherDeviceLoginRequested.link());
|
|
||||||
} else if (clientBoundEvent instanceof OnPasswordRequested onPasswordRequested) {
|
|
||||||
dataOutputStream.writeByte(0x6);
|
|
||||||
dataOutputStream.writeUTF(onPasswordRequested.passwordHint());
|
|
||||||
dataOutputStream.writeBoolean(onPasswordRequested.hasRecoveryEmail());
|
|
||||||
dataOutputStream.writeUTF(onPasswordRequested.recoveryEmailPattern());
|
|
||||||
} else {
|
|
||||||
throw new UnsupportedOperationException("Unexpected value: " + clientBoundEvent);
|
|
||||||
}
|
}
|
||||||
return byteArrayOutputStream.toByteArray();
|
return byteArrayOutputStream.toByteArray();
|
||||||
}
|
}
|
||||||
@ -387,6 +369,46 @@ public abstract class ReactiveApiPublisher {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static byte[] serializeEvent(ClientBoundEvent clientBoundEvent) {
|
||||||
|
try (var byteArrayOutputStream = new ByteArrayOutputStream()) {
|
||||||
|
try (var dataOutputStream = new DataOutputStream(byteArrayOutputStream)) {
|
||||||
|
writeClientBoundEvent(clientBoundEvent, dataOutputStream);
|
||||||
|
return byteArrayOutputStream.toByteArray();
|
||||||
|
}
|
||||||
|
} catch (IOException ex) {
|
||||||
|
throw new SerializationException(ex);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void writeClientBoundEvent(ClientBoundEvent clientBoundEvent, DataOutputStream dataOutputStream)
|
||||||
|
throws IOException {
|
||||||
|
dataOutputStream.writeLong(clientBoundEvent.liveId());
|
||||||
|
dataOutputStream.writeLong(clientBoundEvent.userId());
|
||||||
|
if (clientBoundEvent instanceof OnUpdateData onUpdateData) {
|
||||||
|
dataOutputStream.writeByte(0x1);
|
||||||
|
onUpdateData.update().serialize(dataOutputStream);
|
||||||
|
} else if (clientBoundEvent instanceof OnUpdateError onUpdateError) {
|
||||||
|
dataOutputStream.writeByte(0x2);
|
||||||
|
onUpdateError.error().serialize(dataOutputStream);
|
||||||
|
} else if (clientBoundEvent instanceof OnUserLoginCodeRequested onUserLoginCodeRequested) {
|
||||||
|
dataOutputStream.writeByte(0x3);
|
||||||
|
dataOutputStream.writeLong(onUserLoginCodeRequested.phoneNumber());
|
||||||
|
} else if (clientBoundEvent instanceof OnBotLoginCodeRequested onBotLoginCodeRequested) {
|
||||||
|
dataOutputStream.writeByte(0x4);
|
||||||
|
dataOutputStream.writeUTF(onBotLoginCodeRequested.token());
|
||||||
|
} else if (clientBoundEvent instanceof OnOtherDeviceLoginRequested onOtherDeviceLoginRequested) {
|
||||||
|
dataOutputStream.writeByte(0x5);
|
||||||
|
dataOutputStream.writeUTF(onOtherDeviceLoginRequested.link());
|
||||||
|
} else if (clientBoundEvent instanceof OnPasswordRequested onPasswordRequested) {
|
||||||
|
dataOutputStream.writeByte(0x6);
|
||||||
|
dataOutputStream.writeUTF(onPasswordRequested.passwordHint());
|
||||||
|
dataOutputStream.writeBoolean(onPasswordRequested.hasRecoveryEmail());
|
||||||
|
dataOutputStream.writeUTF(onPasswordRequested.recoveryEmailPattern());
|
||||||
|
} else {
|
||||||
|
throw new UnsupportedOperationException("Unexpected value: " + clientBoundEvent);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private CompletableFuture<Subscription> registerTopics() {
|
private CompletableFuture<Subscription> registerTopics() {
|
||||||
// Start receiving requests
|
// Start receiving requests
|
||||||
eventService.subscribe("session-" + liveId + "-requests",
|
eventService.subscribe("session-" + liveId + "-requests",
|
||||||
|
Loading…
x
Reference in New Issue
Block a user