Skip to content

Deduplication #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
picolino merged 4 commits into from
Mar 7, 2023
Merged

Deduplication #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c7562ca
Add messages deduplication
picolino Mar 7, 2023
27e03d1
Override message hashing while deduplication
picolino Mar 7, 2023
6047bda
Handle remote endpoint disposion exception
picolino Mar 7, 2023
01afe15
Close #2. Separate local topology and network topology
picolino Mar 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 56 additions & 44 deletions src/Chat.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ module Chat =
AppSettingsFilePath: string
UserSettingsFilePath: string
ChatScrollViewOffset: float
MessagesListHashSet: Set<int>
}

type Msg =
Expand Down Expand Up @@ -248,6 +249,7 @@ module Chat =
ConnectedUsers = []
MessageInput = ""
MessagesList = []
MessagesListHashSet = Set.empty
SettingsVisible = false
AppSettingsFilePath = appSettingsFilePath
UserSettingsFilePath = userSettingsFilePath
Expand Down Expand Up @@ -365,41 +367,44 @@ module Chat =
model, Cmd.OfAsync.perform connectToLocalPeers () PeersConnected
| StartConnectToRemotePeersLoop ->
let connectToRemotePeers _ = async {

logger.Debug $"[StartConnectToRemotePeersLoop] Defining non-connected known peers..."

let nonConnectedRemotePeers =
model.KnownPeers
|> List.where (fun ep -> model.Connections |> List.exists (fun x -> x.ConnectionId = ep.ToString()) |> not)
if model.TcpListener.IsBound
then
logger.Debug $"[StartConnectToRemotePeersLoop] Defining non-connected known peers..."

if nonConnectedRemotePeers |> List.length > 0
then logger.Debug $"[StartConnectToRemotePeersLoop] Non-connected known peers found {nonConnectedRemotePeers}. Connecting..."
else logger.Debug $"[StartConnectToRemotePeersLoop] All known peers already connected. Skipping..."

return
nonConnectedRemotePeers
|> Array.ofList
|> Array.Parallel.map (fun ep ->
let socket = Tcp.client model.ClientPort

logger.Debug $"[StartConnectToRemotePeersLoop] Connecting to remote tcp endpoint {ep} from {socket.LocalEndPoint}..."
let nonConnectedRemotePeers =
model.KnownPeers
|> List.where (fun ep -> model.Connections |> List.exists (fun x -> x.ConnectionId = ep.ToString()) |> not)

try
let connectedSocket = Tcp.connectSocket ep.Address ep.Port socket
let connectionEndpoint = {
ConnectionId = ep.ToString()
EndPoint = ep
Client = connectedSocket
}
Some connectionEndpoint
with
| e ->
logger.DebugException e $"[StartConnectToRemotePeersLoop] Connection to remote tcp endpoint {ep} failed"
socket.Dispose()
None
)
|> Array.choose id
|> List.ofArray
if nonConnectedRemotePeers |> List.length > 0
then logger.Debug $"[StartConnectToRemotePeersLoop] Non-connected known peers found {nonConnectedRemotePeers}. Connecting..."
else logger.Debug $"[StartConnectToRemotePeersLoop] All known peers already connected. Skipping..."

return
nonConnectedRemotePeers
|> Array.ofList
|> Array.Parallel.map (fun ep ->
let socket = Tcp.client model.ClientPort

logger.Debug $"[StartConnectToRemotePeersLoop] Connecting to remote tcp endpoint {ep} from {socket.LocalEndPoint}..."

try
let connectedSocket = Tcp.connectSocket ep.Address ep.Port socket
let connectionEndpoint = {
ConnectionId = ep.ToString()
EndPoint = ep
Client = connectedSocket
}
Some connectionEndpoint
with
| e ->
logger.DebugException e $"[StartConnectToRemotePeersLoop] Connection to remote tcp endpoint {ep} failed"
socket.Dispose()
None
)
|> Array.choose id
|> List.ofArray
else
return []
}

model, Cmd.OfAsync.perform connectToRemotePeers () ConnectToRemotePeersIterationFinished
Expand Down Expand Up @@ -434,7 +439,7 @@ module Chat =
| DisconnectClient socket ->
let connections =
model.Connections
|> List.where (fun o -> o.ConnectionId <> socket.RemoteEndPoint.ToString())
|> List.where (fun o -> socket.RemoteEndPoint <> null && o.ConnectionId <> socket.RemoteEndPoint.ToString())

socket.Dispose()

Expand Down Expand Up @@ -539,25 +544,26 @@ module Chat =

logger.Debug $"[RemoteChatMessageReceived] Chat message {msg} package received by {client.RemoteEndPoint}"

match msg.RetranslationInfo.RetranslatedBy |> List.contains model.UserId with
| false ->

let retranslatedByCurrentInstance = msg.RetranslationInfo.RetranslatedBy |> List.contains model.UserId
let receivedMessageAlreadyPresentsHere = model.MessagesListHashSet |> Set.contains (msg.GetHalfHashCode())

if retranslatedByCurrentInstance || receivedMessageAlreadyPresentsHere
then
model, Cmd.none
else
logger.Debug $"[RemoteChatMessageReceived] Chat message {msg} by {client.RemoteEndPoint} not being retranslated by this app. Retranslating..."

match msg.SecretCode = model.SecretCode with
| true ->

if msg.SecretCode = model.SecretCode
then
logger.Info $"[RemoteChatMessageReceived] Chat message {msg} for me by {client.RemoteEndPoint}. Append message to messages list..."

let cmds = [
Cmd.ofMsg <| AppendLocalMessage { Message = msg; IsMe = false }
Cmd.ofMsg <| RetranslateChatMessage msg
]
model, Cmd.batch cmds
| false ->
else
model, Cmd.ofMsg <| RetranslateChatMessage msg
| true ->
model, Cmd.none
| RetranslateChatMessage msg ->
let msg = { msg with RetranslationInfo = { msg.RetranslationInfo with RetranslatedBy = model.UserId :: msg.RetranslationInfo.RetranslatedBy } }
model.Connections
Expand Down Expand Up @@ -587,7 +593,13 @@ module Chat =
| AppendLocalMessage m ->
let msg = WaitThenSend (0, ScrollChatToEnd) // Delay 0 because rendering pipeline can't update vertical offset correctly without async message between frames
let verticalOffset = Double.MaxValue // Vertical offset must be different between frames because of caching inside Avalonia.FuncUI library
{ model with MessagesList = model.MessagesList @ [m]; ChatScrollViewOffset = verticalOffset }, Cmd.ofMsg msg
let model = {
model with
MessagesList = model.MessagesList @ [m]
MessagesListHashSet = model.MessagesListHashSet |> Set.add (m.Message.GetHalfHashCode())
ChatScrollViewOffset = verticalOffset
}
model, Cmd.ofMsg msg
| ScrollChatToEnd ->
{ model with ChatScrollViewOffset = Double.PositiveInfinity }, Cmd.none
| TextChanged t ->
Expand Down
7 changes: 5 additions & 2 deletions src/Message.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,15 @@ module Message =
RetranslationInfo: RetranslationInfo
}

type ChatMessage = {
type ChatMessage =
{
MessageText: string
MessageSender: string
DateTime: DateTime
SecretCode: int
UserId: string
RetranslationInfo: RetranslationInfo
}
}

member this.GetHalfHashCode() = hash <| this.MessageText + this.UserId + this.DateTime.Ticks.ToString() + this.SecretCode.ToString()