dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-28 02:54:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master-ghc9'

Browse Source
This commit is contained in:
Evgeny Poberezkin
2023-09-18 21:18:45 +01:00
parent 00b48babf3 fee897f367
commit 81385e39bf
48 changed files with 354 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/build.yml
+4 -4
View File
@@ -23,13 +23,13 @@ jobs:
platform_name: 22_04-x86-64
steps:
- name: Clone project
uses: actions/checkout@v2
uses: actions/checkout@v3
- name: Setup Haskell
uses: haskell/actions/setup@v1
uses: haskell-actions/setup@v2
with:
ghc-version: "8.10.7"
cabal-version: "latest"
ghc-version: "9.6.2"
cabal-version: "3.10.1.0"
- name: Cache dependencies
uses: actions/cache@v2
Dockerfile
+3 -3
View File
@@ -8,8 +8,8 @@ FROM ubuntu:${TAG} AS build
RUN apt-get update && apt-get install -y curl git build-essential libgmp3-dev zlib1g-dev llvm-12 llvm-12-dev libnuma-dev
# Specify bootstrap Haskell versions
ENV BOOTSTRAP_HASKELL_GHC_VERSION=8.10.7
ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.6.2.0
ENV BOOTSTRAP_HASKELL_GHC_VERSION=9.6.2
ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.10.1.0
# Install ghcup
RUN curl --proto '=https' --tlsv1.2 -sSf https://get-ghcup.haskell.org | BOOTSTRAP_HASKELL_NONINTERACTIVE=1 sh
@@ -19,7 +19,7 @@ ENV PATH="/root/.cabal/bin:/root/.ghcup/bin:$PATH"
# Set both as default
RUN ghcup set ghc "${BOOTSTRAP_HASKELL_GHC_VERSION}" && \
ghcup set cabal
ghcup set cabal "${BOOTSTRAP_HASKELL_CABAL_VERSION}"
COPY . /project
WORKDIR /project
cabal.project
+11 -5
View File
@@ -1,14 +1,15 @@
with-compiler: ghc-8.10.7
packages: .
-- packages: . ../direct-sqlcipher ../sqlcipher-simple
-- packages: . ../hs-socks
-- packages: . ../http2
-- packages: . ../network-transport
with-compiler: ghc-9.6.2
source-repository-package
type: git
location: https://github.com/simplex-chat/aeson.git
tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7
tag: 68330dce8208173c6acf5f62b23acb500ab5d873
source-repository-package
type: git
@@ -23,9 +24,14 @@ source-repository-package
source-repository-package
type: git
location: https://github.com/simplex-chat/direct-sqlcipher.git
tag: 34309410eb2069b029b8fc1872deb1e0db123294
tag: f814ee68b16a9447fbb467ccc8f29bdd3546bfd9
source-repository-package
type: git
location: https://github.com/simplex-chat/sqlcipher-simple.git
tag: 5e154a2aeccc33ead6c243ec07195ab673137221
tag: a46bd361a19376c5211f1058908fc0ae6bf42446
source-repository-package
type: git
location: https://github.com/simplex-chat/network-transport.git
tag: 0013798272a683e35ca38d2fdaf480942311fba8
package.yaml
+18 -12
View File
@@ -22,7 +22,7 @@ extra-source-files:
- CHANGELOG.md
dependencies:
- aeson == 2.0.*
- aeson == 2.2.*
- ansi-terminal >= 0.10 && < 0.12
- asn1-encoding == 0.9.*
- asn1-types == 0.3.*
@@ -30,12 +30,12 @@ dependencies:
- attoparsec == 0.14.*
- base >= 4.14 && < 5
- base64-bytestring >= 1.0 && < 1.3
- bytestring == 0.10.*
- bytestring == 0.11.*
- case-insensitive == 1.2.*
- composition == 1.0.*
- constraints >= 0.12 && < 0.14
- containers == 0.6.*
- cryptonite >= 0.27 && < 0.30
- cryptonite == 0.30.*
- cryptostore == 0.2.*
- data-default == 0.7.*
- direct-sqlcipher == 2.3.*
@@ -43,14 +43,14 @@ dependencies:
- filepath == 1.4.*
- http-types == 0.12.*
- http2 == 4.1.*
- generic-random >= 1.3 && < 1.5
- generic-random == 1.5.*
- ini == 0.4.1
- iproute == 1.7.*
- iso8601-time == 0.1.*
- memory == 0.15.*
- mtl == 2.2.*
- memory == 0.18.*
- mtl == 2.3.*
- network >= 3.1.2.7 && < 3.2
- network-transport == 0.5.4
- network-transport == 0.5.6
- optparse-applicative >= 0.15 && < 0.17
- QuickCheck == 2.14.*
- process == 1.6.*
@@ -59,14 +59,15 @@ dependencies:
- socks == 0.6.*
- sqlcipher-simple == 0.4.*
- stm == 2.5.*
- template-haskell == 2.16.*
- template-haskell == 2.20.*
- temporary == 1.3.*
- text == 1.2.*
- text == 2.0.*
- time == 1.9.*
- time-compat == 1.9.*
- time-manager == 0.0.*
- tls >= 1.6.0 && < 1.7
- transformers == 0.5.*
- transformers == 0.6.*
- unix == 2.8.1.1
- unliftio == 0.2.*
- unliftio-core == 0.2.*
- websockets == 0.12.*
@@ -97,6 +98,7 @@ executables:
- simplexmq
ghc-options:
- -threaded
- -rtsopts
ntf-server:
source-dirs: apps/ntf-server
@@ -105,6 +107,7 @@ executables:
- simplexmq
ghc-options:
- -threaded
- -rtsopts
xftp-server:
source-dirs: apps/xftp-server
@@ -113,6 +116,7 @@ executables:
- simplexmq
ghc-options:
- -threaded
- -rtsopts
smp-agent:
source-dirs: apps/smp-agent
@@ -121,6 +125,7 @@ executables:
- simplexmq
ghc-options:
- -threaded
- -rtsopts
xftp:
source-dirs: apps/xftp
@@ -129,6 +134,7 @@ executables:
- simplexmq
ghc-options:
- -threaded
- -rtsopts
tests:
simplexmq-test:
@@ -137,8 +143,8 @@ tests:
dependencies:
- simplexmq
- deepseq == 1.4.*
- hspec == 2.7.*
- hspec-core == 2.7.*
- hspec == 2.11.*
- hspec-core == 2.11.*
- HUnit == 1.6.*
- QuickCheck == 2.14.*
- silently == 1.2.*
simplexmq.cabal
+85 -78
View File
@@ -1,6 +1,6 @@
cabal-version: 1.12
-- This file has been generated from package.yaml by hpack version 0.35.1.
-- This file has been generated from package.yaml by hpack version 0.35.2.
--
-- see: https://github.com/sol/hpack
@@ -145,7 +145,7 @@ library
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
build-depends:
QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -153,27 +153,27 @@ library
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, generic-random ==1.5.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -181,14 +181,15 @@ library
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
@@ -206,10 +207,10 @@ executable ntf-server
Paths_simplexmq
hs-source-dirs:
apps/ntf-server
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
build-depends:
QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -217,27 +218,27 @@ executable ntf-server
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, generic-random ==1.5.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -246,14 +247,15 @@ executable ntf-server
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
@@ -271,10 +273,10 @@ executable smp-agent
Paths_simplexmq
hs-source-dirs:
apps/smp-agent
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
build-depends:
QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -282,27 +284,27 @@ executable smp-agent
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, generic-random ==1.5.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -311,14 +313,15 @@ executable smp-agent
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
@@ -336,10 +339,10 @@ executable smp-server
Paths_simplexmq
hs-source-dirs:
apps/smp-server
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
build-depends:
QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -347,27 +350,27 @@ executable smp-server
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, generic-random ==1.5.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -376,14 +379,15 @@ executable smp-server
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
@@ -401,10 +405,10 @@ executable xftp
Paths_simplexmq
hs-source-dirs:
apps/xftp
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
build-depends:
QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -412,27 +416,27 @@ executable xftp
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, generic-random ==1.5.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -441,14 +445,15 @@ executable xftp
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
@@ -466,10 +471,10 @@ executable xftp-server
Paths_simplexmq
hs-source-dirs:
apps/xftp-server
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
build-depends:
QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -477,27 +482,27 @@ executable xftp-server
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, generic-random ==1.5.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -506,14 +511,15 @@ executable xftp-server
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
@@ -563,7 +569,7 @@ test-suite simplexmq-test
build-depends:
HUnit ==1.6.*
, QuickCheck ==2.14.*
, aeson ==2.0.*
, aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
, asn1-encoding ==0.9.*
, asn1-types ==0.3.*
@@ -571,31 +577,31 @@ test-suite simplexmq-test
, attoparsec ==0.14.*
, base >=4.14 && <5
, base64-bytestring >=1.0 && <1.3
, bytestring ==0.10.*
, bytestring ==0.11.*
, case-insensitive ==1.2.*
, composition ==1.0.*
, constraints >=0.12 && <0.14
, containers ==0.6.*
, cryptonite >=0.27 && <0.30
, cryptonite ==0.30.*
, cryptostore ==0.2.*
, data-default ==0.7.*
, deepseq ==1.4.*
, direct-sqlcipher ==2.3.*
, directory ==1.3.*
, filepath ==1.4.*
, generic-random >=1.3 && <1.5
, hspec ==2.7.*
, hspec-core ==2.7.*
, generic-random ==1.5.*
, hspec ==2.11.*
, hspec-core ==2.11.*
, http-types ==0.12.*
, http2 ==4.1.*
, ini ==0.4.1
, iproute ==1.7.*
, iso8601-time ==0.1.*
, main-tester ==0.2.*
, memory ==0.15.*
, mtl ==2.2.*
, memory ==0.18.*
, mtl ==2.3.*
, network >=3.1.2.7 && <3.2
, network-transport ==0.5.4
, network-transport ==0.5.6
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, random >=1.1 && <1.3
@@ -605,15 +611,16 @@ test-suite simplexmq-test
, socks ==0.6.*
, sqlcipher-simple ==0.4.*
, stm ==2.5.*
, template-haskell ==2.16.*
, template-haskell ==2.20.*
, temporary ==1.3.*
, text ==1.2.*
, text ==2.0.*
, time ==1.9.*
, time-compat ==1.9.*
, time-manager ==0.0.*
, timeit ==2.0.*
, tls >=1.6.0 && <1.7
, transformers ==0.5.*
, transformers ==0.6.*
, unix ==2.8.1.1
, unliftio ==0.2.*
, unliftio-core ==0.2.*
, websockets ==0.12.*
src/Simplex/FileTransfer/Agent.hs
+4 -2
View File
@@ -4,12 +4,15 @@
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeApplications #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module Simplex.FileTransfer.Agent
( startXFTPWorkers,
closeXFTPAgent,
@@ -24,7 +27,6 @@ module Simplex.FileTransfer.Agent
)
where
import Control.Concurrent.STM (stateTVar)
import Control.Logger.Simple (logError)
import Control.Monad
import Control.Monad.Except
@@ -486,7 +488,7 @@ runXFTPSndWorker c srv doWork = do
rcvChunks :: [[FileChunk]]
rcvChunks = map (sortChunks . M.elems) $ M.elems $ foldl' addRcvChunk M.empty rcvReplicas
sortChunks :: [FileChunk] -> [FileChunk]
sortChunks = map reverseReplicas . sortOn (chunkNo :: FileChunk -> Int)
sortChunks = map reverseReplicas . sortOn (\fc -> fc.chunkNo)
reverseReplicas ch@FileChunk {replicas} = (ch :: FileChunk) {replicas = reverse replicas}
addRcvChunk :: Map Int (Map Int FileChunk) -> SentRecipientReplica -> Map Int (Map Int FileChunk)
addRcvChunk m SentRecipientReplica {chunkNo, server, rcvNo, replicaId, replicaKey, digest, chunkSize} =
src/Simplex/FileTransfer/Client.hs
+3 -1
View File
@@ -3,12 +3,14 @@
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications #-}
module Simplex.FileTransfer.Client where
import Control.Monad
import Control.Monad.Except
import Data.Bifunctor (first)
import Data.ByteString.Builder (Builder, byteString)
@@ -108,7 +110,7 @@ xftpClientServer = B.unpack . strEncode . snd3 . transportSession
snd3 (_, s, _) = s
xftpTransportHost :: XFTPClient -> TransportHost
xftpTransportHost = (host :: HClient -> TransportHost) . client_ . http2Client
xftpTransportHost c = c.http2Client.client_.host
xftpSessionTs :: XFTPClient -> UTCTime
xftpSessionTs = sessionTs . http2Client
src/Simplex/FileTransfer/Client/Agent.hs
+2 -1
View File
@@ -8,6 +8,7 @@
module Simplex.FileTransfer.Client.Agent where
import Control.Logger.Simple (logInfo)
import Control.Monad
import Control.Monad.Except
import Data.Bifunctor (first)
import qualified Data.ByteString.Char8 as B
@@ -21,7 +22,7 @@ import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (ProtocolServer (..), XFTPServer)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util (catchAll_, tryError)
import Simplex.Messaging.Util (catchAll_)
import UnliftIO
type XFTPClientVar = TMVar (Either XFTPClientAgentError XFTPClient)
src/Simplex/FileTransfer/Client/Main.hs
+6 -4
View File
@@ -5,10 +5,13 @@
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module Simplex.FileTransfer.Client.Main
( SendOptions (..),
CLIError (..),
@@ -24,7 +27,6 @@ module Simplex.FileTransfer.Client.Main
)
where
import Control.Concurrent.STM (stateTVar)
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
@@ -360,7 +362,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
rcvChunks :: [[FileChunk]]
rcvChunks = map (sortChunks . M.elems) $ M.elems $ foldl' addRcvChunk M.empty rcvReplicas
sortChunks :: [FileChunk] -> [FileChunk]
sortChunks = map reverseReplicas . sortOn (chunkNo :: FileChunk -> Int)
sortChunks = map reverseReplicas . sortOn (\c -> c.chunkNo)
reverseReplicas ch@FileChunk {replicas} = (ch :: FileChunk) {replicas = reverse replicas}
addRcvChunk :: Map Int (Map Int FileChunk) -> SentRecipientReplica -> Map Int (Map Int FileChunk)
addRcvChunk m SentRecipientReplica {chunkNo, server, rcvNo, replicaId, replicaKey, digest, chunkSize} =
@@ -419,7 +421,7 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
a <- atomically $ newXFTPAgent defaultXFTPClientAgentConfig
liftIO $ printNoNewLine "Downloading file..."
downloadedChunks <- newTVarIO []
let srv FileChunk {replicas} = server (head replicas :: FileChunkReplica)
let srv FileChunk {replicas} = (head replicas).server
srvChunks = groupAllOn srv chunks
chunkPaths <- map snd . sortOn fst . concat <$> pooledForConcurrentlyN 16 srvChunks (mapM $ downloadFileChunk a encPath size downloadedChunks)
encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
@@ -498,7 +500,7 @@ cliFileDescrInfo InfoOptions {fileDescription} = do
putStrLn $ "File download size: " <> strEnc size
putStrLn "File server(s):"
forM_ replicas $ \srvReplicas -> do
let srv = replicaServer $ head srvReplicas
let srv = (head srvReplicas).server
chSizes = map (\FileServerReplica {chunkSize = chSize_} -> unFileSize $ fromMaybe chunkSize chSize_) srvReplicas
putStrLn $ strEnc srv <> ": " <> strEnc (FileSize $ sum chSizes)
where
src/Simplex/FileTransfer/Crypto.hs
+1
View File
@@ -6,6 +6,7 @@
module Simplex.FileTransfer.Crypto where
import Control.Monad
import Control.Monad.Except
import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.Bifunctor (first)
src/Simplex/FileTransfer/Description.hs
+7 -8
View File
@@ -5,10 +5,13 @@
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module Simplex.FileTransfer.Description
( FileDescription (..),
AFileDescription (..),
@@ -25,7 +28,6 @@ module Simplex.FileTransfer.Description
YAMLServerReplicas (..), -- for tests
validateFileDescription,
groupReplicasByServer,
replicaServer,
fdSeparator,
kb,
mb,
@@ -200,7 +202,7 @@ validateFileDescription fd@FileDescription {size, chunks}
| chunksSize chunks /= unFileSize size = Left "chunks total size is different than file size"
| otherwise = Right $ ValidFD fd
where
chunkNos = map (chunkNo :: FileChunk -> Int) chunks
chunkNos = map (\c -> c.chunkNo) chunks
chunksSize = fromIntegral . foldl' (\s FileChunk {chunkSize} -> s + unFileSize chunkSize) 0
encodeFileDescription :: FileDescription p -> YAMLFileDescription
@@ -246,7 +248,7 @@ instance (ToField a) => ToField (FileSize a) where toField (FileSize s) = toFiel
groupReplicasByServer :: FileSize Word32 -> [FileChunk] -> [[FileServerReplica]]
groupReplicasByServer defChunkSize =
groupAllOn replicaServer . unfoldChunksToReplicas defChunkSize
groupAllOn (\r -> r.server) . unfoldChunksToReplicas defChunkSize
encodeFileReplicas :: FileSize Word32 -> [FileChunk] -> [YAMLServerReplicas]
encodeFileReplicas defChunkSize =
@@ -254,13 +256,10 @@ encodeFileReplicas defChunkSize =
where
encodeServerReplicas fs =
YAMLServerReplicas
{ server = replicaServer $ head fs, -- groupAllOn guarantees that fs is not empty
{ server = (head fs).server, -- groupAllOn guarantees that fs is not empty
chunks = map (B.unpack . encodeServerReplica) fs
}
replicaServer :: FileServerReplica -> XFTPServer
replicaServer = server
encodeServerReplica :: FileServerReplica -> ByteString
encodeServerReplica FileServerReplica {chunkNo, replicaId, replicaKey, digest, chunkSize} =
bshow chunkNo
@@ -312,7 +311,7 @@ foldReplicasToChunks :: FileSize Word32 -> [FileServerReplica] -> Either String
foldReplicasToChunks defChunkSize fs = do
sd <- foldSizesDigests fs
-- TODO validate (check that chunks match) or in separate function
sortOn (chunkNo :: FileChunk -> Int) . map reverseReplicas . M.elems <$> foldChunks sd fs
sortOn (\c -> c.chunkNo) . map reverseReplicas . M.elems <$> foldChunks sd fs
where
foldSizesDigests :: [FileServerReplica] -> Either String (Map Int (FileSize Word32), Map Int FileDigest)
foldSizesDigests = foldl' addSizeDigest $ Right (M.empty, M.empty)
src/Simplex/FileTransfer/Server.hs
+1
View File
@@ -12,6 +12,7 @@
module Simplex.FileTransfer.Server where
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Unlift (MonadUnliftIO)
import Control.Monad.Reader
src/Simplex/FileTransfer/Transport.hs
+2
View File
@@ -16,7 +16,9 @@ module Simplex.FileTransfer.Transport
where
import qualified Control.Exception as E
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Class
import qualified Data.ByteArray as BA
import Data.ByteString.Builder (Builder, byteString)
import Data.ByteString.Char8 (ByteString)
src/Simplex/Messaging/Agent.hs
+12 -9
View File
@@ -8,12 +8,15 @@
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeApplications #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
-- |
-- Module : Simplex.Messaging.Agent
-- Copyright : (c) simplex.chat
@@ -102,8 +105,8 @@ module Simplex.Messaging.Agent
)
where
import Control.Concurrent.STM (stateTVar)
import Control.Logger.Simple (logError, logInfo, showText)
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Unlift (MonadUnliftIO)
import Control.Monad.Reader
@@ -734,7 +737,7 @@ subscribeConnections' c connIds = do
ContactConnection _ rq -> Right [rq]
NewConnection _ -> Left (Right ())
sndSubResult :: SndQueue -> Either AgentErrorType ()
sndSubResult sq = case status (sq :: SndQueue) of
sndSubResult SndQueue {status} = case status of
Confirmed -> Right ()
Active -> Left $ CONN SIMPLEX
_ -> Left $ INTERNAL "unexpected queue status"
@@ -941,7 +944,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) ->
case find (sameQueue (srv, rId)) rqs of
Just rq'@RcvQueue {server, sndId, status, dbReplaceQueueId = Just replaceQId} ->
case find ((replaceQId ==) . dbQId) rqs of
case find (\q -> replaceQId == q.dbQueueId) rqs of
Just rq1 -> when (status == Confirmed) $ do
secureQueue c rq' senderKey
withStore' c $ \db -> setRcvQueueStatus db rq' Secured
@@ -1198,7 +1201,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
-- this is the same queue where this loop delivers messages to but with updated state
Just SndQueue {dbReplaceQueueId = Just replacedId, primary} ->
-- second part of this condition is a sanity check because dbReplaceQueueId cannot point to the same queue, see switchConnection'
case removeQP (\sq' -> dbQId sq' == replacedId && not (sameQueue addr sq')) sqs of
case removeQP (\sq' -> sq'.dbQueueId == replacedId && not (sameQueue addr sq')) sqs of
Nothing -> internalErr msgId "sent QTEST: queue not found in connection"
Just (sq', sq'' : sqs') -> do
checkSQSwchStatus sq' SSSendingQTEST
@@ -1313,7 +1316,7 @@ abortConnectionSwitch' c connId =
| canAbortRcvSwitch rq -> do
when (ratchetSyncSendProhibited cData) $ throwError $ CMD PROHIBITED
-- multiple queues to which the connections switches were possible when repeating switch was allowed
let (delRqs, keepRqs) = L.partition ((Just (dbQId rq) ==) . dbReplaceQId) rqs
let (delRqs, keepRqs) = L.partition (\q -> Just rq.dbQueueId == q.dbReplaceQueueId) rqs
case L.nonEmpty keepRqs of
Just rqs' -> do
rq' <- withStore' c $ \db -> do
@@ -1431,7 +1434,7 @@ deleteConnQueues c ntf rqs = do
| temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> withStore' c (`incRcvDeleteErrors` rq) $> r
| otherwise -> withStore' c (`deleteConnRcvQueue` rq) >> notifyRQ rq (Just e) $> Right ()
pure (rq, r')
notifyRQ rq e_ = notify ("", qConnId rq, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
notifyRQ rq e_ = notify ("", rq.connId, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
notify = when ntf . atomically . writeTBQueue (subQ c)
connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
connResults = M.map snd . foldl' addResult M.empty
@@ -1876,7 +1879,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
case (conn', dbReplaceQueueId) of
(DuplexConnection _ rqs _, Just replacedId) -> do
when primary . withStore' c $ \db -> setRcvQueuePrimary db connId rq
case find ((replacedId ==) . dbQId) rqs of
case find (\q -> replacedId == q.dbQueueId) rqs of
Just rq'@RcvQueue {server, rcvId} -> do
checkRQSwchStatus rq' RSSendingQUSE
void $ withStore' c $ \db -> setRcvSwitchStatus db rq' $ Just RSReceivedMessage
@@ -2151,7 +2154,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
case (findQ (qAddress sqInfo) sqs, findQ addr sqs) of
(Just _, _) -> qError "QADD: queue address is already used in connection"
(_, Just sq@SndQueue {dbQueueId}) -> do
let (delSqs, keepSqs) = L.partition ((Just dbQueueId ==) . dbReplaceQId) sqs
let (delSqs, keepSqs) = L.partition (\q -> Just dbQueueId == q.dbReplaceQueueId) sqs
case L.nonEmpty keepSqs of
Just sqs' -> do
-- move inside case?
@@ -2202,7 +2205,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
case findQ addr sqs of
Just sq'@SndQueue {dbReplaceQueueId = Just replaceQId} -> do
case find ((replaceQId ==) . dbQId) sqs of
case find (\q -> replaceQId == q.dbQueueId) sqs of
Just sq1 -> do
checkSQSwchStatus sq1 SSSendingQKEY
logServer "<--" c srv rId $ "MSG <QUSE> " <> logSecret (snd addr)
src/Simplex/Messaging/Agent/Client.hs
+5 -3
View File
@@ -10,6 +10,7 @@
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
@@ -107,9 +108,10 @@ where
import Control.Applicative ((<|>))
import Control.Concurrent (forkIO, threadDelay)
import Control.Concurrent.Async (Async, uninterruptibleCancel)
import Control.Concurrent.STM (retry, stateTVar, throwSTM)
import Control.Concurrent.STM (retry, throwSTM)
import Control.Exception (AsyncException (..))
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Unlift
import Control.Monad.Reader
@@ -431,7 +433,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
TM.delete tSess smpClients
qs <- RQ.getDelSessQueues tSess $ activeSubs c
mapM_ (`RQ.addQueue` pendingSubs c) qs
let cs = S.fromList $ map qConnId qs
let cs = S.fromList $ map (\q -> q.connId) qs
cs' <- RQ.getConns $ activeSubs c
pure (qs, S.toList $ cs `S.difference` cs')
@@ -819,7 +821,7 @@ mkSMPTransportSession :: (AgentMonad' m, SMPQueueRec q) => AgentClient -> q -> m
mkSMPTransportSession c q = mkSMPTSession q <$> getSessionMode c
mkSMPTSession :: SMPQueueRec q => q -> TransportSessionMode -> SMPTransportSession
mkSMPTSession q = mkTSession (qUserId q) (qServer q) (qConnId q)
mkSMPTSession q = mkTSession q.userId (qServer q) q.connId
getSessionMode :: AgentMonad' m => AgentClient -> m TransportSessionMode
getSessionMode = fmap sessionMode . readTVarIO . useNetworkConfig
src/Simplex/Messaging/Agent/NtfSubSupervisor.hs
+5 -4
View File
@@ -6,6 +6,8 @@
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module Simplex.Messaging.Agent.NtfSubSupervisor
( runNtfSupervisor,
nsUpdateToken,
@@ -17,7 +19,6 @@ module Simplex.Messaging.Agent.NtfSubSupervisor
)
where
import Control.Concurrent.STM (stateTVar)
import Control.Logger.Simple (logError, logInfo)
import Control.Monad
import Control.Monad.Except
@@ -51,13 +52,13 @@ runNtfSupervisor c = do
ns <- asks ntfSupervisor
forever $ do
cmd@(connId, _) <- atomically . readTBQueue $ ntfSubQ ns
handleError connId . agentOperationBracket c AONtfNetwork waitUntilActive $
handleErr connId . agentOperationBracket c AONtfNetwork waitUntilActive $
runExceptT (processNtfSub c cmd) >>= \case
Left e -> notifyErr connId e
Right _ -> return ()
where
handleError :: ConnId -> m () -> m ()
handleError connId = E.handle $ \(e :: E.SomeException) -> do
handleErr :: ConnId -> m () -> m ()
handleErr connId = E.handle $ \(e :: E.SomeException) -> do
logError $ "runNtfSupervisor error " <> tshow e
notifyErr connId e
notifyErr connId e = notifyInternalError c connId $ "runNtfSupervisor error " <> show e
src/Simplex/Messaging/Agent/Protocol.hs
+17 -14
View File
@@ -80,6 +80,8 @@ module Simplex.Messaging.Agent.Protocol
SMPServerWithAuth,
SrvLoc (..),
SMPQueue (..),
qAddress,
sameQueue,
sameQAddress,
noAuthSrv,
SMPQueueUri (..),
@@ -1196,8 +1198,15 @@ updateSMPServerHosts srv@ProtocolServer {host} = case host of
class SMPQueue q where
qServer :: q -> SMPServer
qAddress :: q -> (SMPServer, SMP.QueueId)
sameQueue :: (SMPServer, SMP.QueueId) -> q -> Bool
queueId :: q -> SMP.QueueId
qAddress :: SMPQueue q => q -> (SMPServer, SMP.QueueId)
qAddress q = (qServer q, queueId q)
{-# INLINE qAddress #-}
sameQueue :: SMPQueue q => (SMPServer, SMP.QueueId) -> q -> Bool
sameQueue addr q = sameQAddress addr (qAddress q)
{-# INLINE sameQueue #-}
data SMPQueueInfo = SMPQueueInfo {clientVersion :: Version, queueAddress :: SMPQueueAddress}
deriving (Eq, Show)
@@ -1242,26 +1251,20 @@ data SMPQueueAddress = SMPQueueAddress
instance SMPQueue SMPQueueUri where
qServer SMPQueueUri {queueAddress} = qServer queueAddress
{-# INLINE qServer #-}
qAddress SMPQueueUri {queueAddress} = qAddress queueAddress
{-# INLINE qAddress #-}
sameQueue addr q = sameQAddress addr (qAddress q)
{-# INLINE sameQueue #-}
queueId SMPQueueUri {queueAddress} = queueId queueAddress
{-# INLINE queueId #-}
instance SMPQueue SMPQueueInfo where
qServer SMPQueueInfo {queueAddress} = qServer queueAddress
{-# INLINE qServer #-}
qAddress SMPQueueInfo {queueAddress} = qAddress queueAddress
{-# INLINE qAddress #-}
sameQueue addr q = sameQAddress addr (qAddress q)
{-# INLINE sameQueue #-}
queueId SMPQueueInfo {queueAddress} = queueId queueAddress
{-# INLINE queueId #-}
instance SMPQueue SMPQueueAddress where
qServer SMPQueueAddress {smpServer} = smpServer
{-# INLINE qServer #-}
qAddress SMPQueueAddress {smpServer, senderId} = (smpServer, senderId)
{-# INLINE qAddress #-}
sameQueue addr q = sameQAddress addr (qAddress q)
{-# INLINE sameQueue #-}
queueId SMPQueueAddress {senderId} = senderId
{-# INLINE queueId #-}
sameQAddress :: (SMPServer, SMP.QueueId) -> (SMPServer, SMP.QueueId) -> Bool
sameQAddress (srv, qId) (srv', qId') = sameSrvAddr srv srv' && qId == qId'
src/Simplex/Messaging/Agent/Server.hs
+1 -1
View File
@@ -11,7 +11,7 @@ module Simplex.Messaging.Agent.Server
where
import Control.Logger.Simple (logInfo)
import Control.Monad.Except
import Control.Monad
import Control.Monad.IO.Unlift (MonadUnliftIO)
import Control.Monad.Reader
import Crypto.Random (MonadRandom)
src/Simplex/Messaging/Agent/Store.hs
+16 -40
View File
@@ -1,11 +1,14 @@
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StandaloneDeriving #-}
@@ -24,6 +27,7 @@ import qualified Data.List.NonEmpty as L
import Data.Maybe (isJust)
import Data.Time (UTCTime)
import Data.Type.Equality
import GHC.Records (HasField)
import Simplex.Messaging.Agent.Protocol
import Simplex.Messaging.Agent.RetryInterval (RI2State)
import qualified Simplex.Messaging.Crypto as C
@@ -36,7 +40,6 @@ import Simplex.Messaging.Protocol
NotifierId,
NtfPrivateSignKey,
NtfPublicVerifyKey,
QueueId,
RcvDhSecret,
RcvNtfDhSecret,
RcvPrivateSignKey,
@@ -144,18 +147,14 @@ sndQueueInfo SndQueue {server, sndSwchStatus} =
instance SMPQueue RcvQueue where
qServer RcvQueue {server} = server
{-# INLINE qServer #-}
qAddress RcvQueue {server, rcvId} = (server, rcvId)
{-# INLINE qAddress #-}
sameQueue addr q = sameQAddress addr (qAddress q)
{-# INLINE sameQueue #-}
queueId RcvQueue {rcvId} = rcvId
{-# INLINE queueId #-}
instance SMPQueue SndQueue where
qServer SndQueue {server} = server
{-# INLINE qServer #-}
qAddress SndQueue {server, sndId} = (server, sndId)
{-# INLINE qAddress #-}
sameQueue addr q = sameQAddress addr (qAddress q)
{-# INLINE sameQueue #-}
queueId SndQueue {sndId} = sndId
{-# INLINE queueId #-}
findQ :: SMPQueue q => (SMPServer, SMP.QueueId) -> NonEmpty q -> Maybe q
findQ = find . sameQueue
@@ -183,39 +182,16 @@ switchingRQ = find $ isJust . rcvSwchStatus
{-# INLINE switchingRQ #-}
updatedQs :: SMPQueueRec q => q -> NonEmpty q -> NonEmpty q
updatedQs q = L.map $ \q' -> if dbQId q == dbQId q' then q else q'
updatedQs q = L.map $ \q' -> if q.dbQueueId == q'.dbQueueId then q else q'
{-# INLINE updatedQs #-}
class SMPQueue q => SMPQueueRec q where
qUserId :: q -> UserId
qConnId :: q -> ConnId
queueId :: q -> QueueId
dbQId :: q -> Int64
dbReplaceQId :: q -> Maybe Int64
instance SMPQueueRec RcvQueue where
qUserId = userId
{-# INLINE qUserId #-}
qConnId = connId
{-# INLINE qConnId #-}
queueId = rcvId
{-# INLINE queueId #-}
dbQId = dbQueueId
{-# INLINE dbQId #-}
dbReplaceQId = dbReplaceQueueId
{-# INLINE dbReplaceQId #-}
instance SMPQueueRec SndQueue where
qUserId = userId
{-# INLINE qUserId #-}
qConnId = connId
{-# INLINE qConnId #-}
queueId = sndId
{-# INLINE queueId #-}
dbQId = dbQueueId
{-# INLINE dbQId #-}
dbReplaceQId = dbReplaceQueueId
{-# INLINE dbReplaceQId #-}
type SMPQueueRec q =
( SMPQueue q,
HasField "userId" q UserId,
HasField "connId" q ConnId,
HasField "dbQueueId" q Int64,
HasField "dbReplaceQueueId" q (Maybe Int64)
)
-- * Connection types
src/Simplex/Messaging/Agent/Store/SQLite.hs
+6 -2
View File
@@ -10,6 +10,7 @@
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE QuasiQuotes #-}
@@ -18,7 +19,9 @@
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeOperators #-}
{-# LANGUAGE UndecidableInstances #-}
{-# OPTIONS_GHC -fno-warn-orphans #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module Simplex.Messaging.Agent.Store.SQLite
( SQLiteStore (..),
@@ -211,8 +214,9 @@ module Simplex.Messaging.Agent.Store.SQLite
)
where
import Control.Concurrent.STM (stateTVar)
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Class
import Crypto.Random (ChaChaDRG, randomBytesGenerate)
import Data.Aeson (ToJSON)
import qualified Data.Aeson as J
@@ -1955,7 +1959,7 @@ updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =
WHERE conn_id = :conn_id
AND last_internal_rcv_msg_id = :last_internal_rcv_msg_id;
|]
[ ":last_external_snd_msg_id" := sndMsgId (msgMeta :: MsgMeta),
[ ":last_external_snd_msg_id" := msgMeta.sndMsgId,
":last_rcv_msg_hash" := internalHash,
":conn_id" := connId,
":last_internal_rcv_msg_id" := internalRcvId
src/Simplex/Messaging/Agent/TAsyncs.hs
-1
View File
@@ -1,6 +1,5 @@
module Simplex.Messaging.Agent.TAsyncs where
import Control.Concurrent.STM (stateTVar)
import Control.Monad.IO.Unlift (MonadUnliftIO)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
src/Simplex/Messaging/Client/Agent.hs
+4 -3
View File
@@ -14,6 +14,7 @@ module Simplex.Messaging.Client.Agent where
import Control.Concurrent (forkIO)
import Control.Concurrent.Async (Async, uninterruptibleCancel)
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Unlift
import Control.Monad.Trans.Except
@@ -39,7 +40,7 @@ import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPS
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Transport
import Simplex.Messaging.Util (catchAll_, tryE, ($>>=), toChunks)
import Simplex.Messaging.Util (catchAll_, ($>>=), toChunks)
import System.Timeout (timeout)
import UnliftIO (async)
import UnliftIO.Exception (Exception)
@@ -272,13 +273,13 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE
subscribeQueue :: SMPClientAgent -> SMPServer -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
subscribeQueue ca srv sub = do
atomically $ addPendingSubscription ca srv sub
withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleError
withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleErr
where
subscribe_ smp = do
smpSubscribe smp sub
atomically $ addSubscription ca srv sub
handleError e = do
handleErr e = do
atomically . when (e /= PCENetworkError && e /= PCEResponseTimeout) $
removePendingSubscription ca srv $ fst sub
throwE e
src/Simplex/Messaging/Crypto.hs
+1
View File
@@ -149,6 +149,7 @@ where
import Control.Concurrent.STM
import Control.Exception (Exception)
import Control.Monad
import Control.Monad.Except
import Control.Monad.Trans.Except
import Crypto.Cipher.AES (AES256)
src/Simplex/Messaging/Crypto/File.hs
+2 -2
View File
@@ -21,8 +21,8 @@ module Simplex.Messaging.Crypto.File
)
where
import Control.Concurrent.STM (stateTVar)
import Control.Exception
import Control.Monad
import Control.Monad.Except
import Data.Aeson (FromJSON, ToJSON)
import qualified Data.Aeson as J
@@ -39,7 +39,7 @@ import Simplex.Messaging.Crypto.Lazy (LazyByteString)
import qualified Simplex.Messaging.Crypto.Lazy as LC
import Simplex.Messaging.Util (liftEitherWith)
import System.Directory (getFileSize)
import UnliftIO (Handle, IOMode (..))
import UnliftIO (Handle, IOMode (..), liftIO)
import qualified UnliftIO as IO
import UnliftIO.STM
src/Simplex/Messaging/Crypto/Ratchet.hs
+1 -1
View File
@@ -15,6 +15,7 @@
module Simplex.Messaging.Crypto.Ratchet where
import Control.Monad.Except
import Control.Monad.IO.Class
import Control.Monad.Trans.Except
import Crypto.Cipher.AES (AES256)
import Crypto.Hash (SHA512)
@@ -38,7 +39,6 @@ import Simplex.Messaging.Crypto
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Parsers (blobFieldDecoder, parseE, parseE')
import Simplex.Messaging.Util (tryE)
import Simplex.Messaging.Version
currentE2EEncryptVersion :: Version
src/Simplex/Messaging/Notifications/Server.hs
+1 -1
View File
@@ -12,8 +12,8 @@
module Simplex.Messaging.Notifications.Server where
import Control.Concurrent.STM (stateTVar)
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.Reader
import Data.Bifunctor (second)
src/Simplex/Messaging/Notifications/Server/Push/APNS.hs
+2
View File
@@ -12,7 +12,9 @@ module Simplex.Messaging.Notifications.Server.Push.APNS where
import Control.Exception (Exception)
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Class
import Crypto.Hash.Algorithms (SHA256 (..))
import qualified Crypto.PubKey.ECC.ECDSA as EC
import qualified Crypto.PubKey.ECC.Types as ECT
src/Simplex/Messaging/Notifications/Server/Stats.hs
+21 -20
View File
@@ -1,5 +1,6 @@
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
module Simplex.Messaging.Notifications.Server.Stats where
@@ -56,30 +57,30 @@ newNtfServerStats ts = do
getNtfServerStatsData :: NtfServerStats -> STM NtfServerStatsData
getNtfServerStatsData s = do
_fromTime <- readTVar $ fromTime (s :: NtfServerStats)
_tknCreated <- readTVar $ tknCreated s
_tknVerified <- readTVar $ tknVerified s
_tknDeleted <- readTVar $ tknDeleted s
_subCreated <- readTVar $ subCreated s
_subDeleted <- readTVar $ subDeleted s
_ntfReceived <- readTVar $ ntfReceived s
_ntfDelivered <- readTVar $ ntfDelivered s
_activeTokens <- getPeriodStatsData $ activeTokens s
_activeSubs <- getPeriodStatsData $ activeSubs s
_fromTime <- readTVar s.fromTime
_tknCreated <- readTVar s.tknCreated
_tknVerified <- readTVar s.tknVerified
_tknDeleted <- readTVar s.tknDeleted
_subCreated <- readTVar s.subCreated
_subDeleted <- readTVar s.subDeleted
_ntfReceived <- readTVar s.ntfReceived
_ntfDelivered <- readTVar s.ntfDelivered
_activeTokens <- getPeriodStatsData s.activeTokens
_activeSubs <- getPeriodStatsData s.activeSubs
pure NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs}
setNtfServerStats :: NtfServerStats -> NtfServerStatsData -> STM ()
setNtfServerStats s d = do
writeTVar (fromTime (s :: NtfServerStats)) $! _fromTime (d :: NtfServerStatsData)
writeTVar (tknCreated s) $! _tknCreated d
writeTVar (tknVerified s) $! _tknVerified d
writeTVar (tknDeleted s) $! _tknDeleted d
writeTVar (subCreated s) $! _subCreated d
writeTVar (subDeleted s) $! _subDeleted d
writeTVar (ntfReceived s) $! _ntfReceived d
writeTVar (ntfDelivered s) $! _ntfDelivered d
setPeriodStats (activeTokens s) (_activeTokens d)
setPeriodStats (activeSubs s) (_activeSubs d)
writeTVar s.fromTime $! d._fromTime
writeTVar s.tknCreated $! _tknCreated d
writeTVar s.tknVerified $! _tknVerified d
writeTVar s.tknDeleted $! _tknDeleted d
writeTVar s.subCreated $! _subCreated d
writeTVar s.subDeleted $! _subDeleted d
writeTVar s.ntfReceived $! _ntfReceived d
writeTVar s.ntfDelivered $! _ntfDelivered d
setPeriodStats s.activeTokens (_activeTokens d)
setPeriodStats s.activeSubs (_activeSubs d)
instance StrEncoding NtfServerStatsData where
strEncode NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs} =
src/Simplex/Messaging/Notifications/Server/StoreLog.hs
+2
View File
@@ -6,6 +6,8 @@
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE StrictData #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module Simplex.Messaging.Notifications.Server.StoreLog
( StoreLog,
NtfStoreLogRecord (..),
src/Simplex/Messaging/Server.hs
+64 -27
View File
@@ -1,5 +1,6 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
@@ -48,16 +49,19 @@ import qualified Data.ByteString.Char8 as B
import Data.Either (fromRight, partitionEithers)
import Data.Functor (($>))
import Data.Int (Int64)
import Data.List (intercalate)
import Data.List (intercalate, sort)
import qualified Data.List.NonEmpty as L
import qualified Data.Map.Strict as M
import Data.Maybe (isNothing)
import Data.Maybe (fromMaybe, isNothing)
import qualified Data.Text as T
import Data.Text.Encoding (decodeLatin1)
import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
import Data.Time.Clock.System (SystemTime (..), getSystemTime)
import Data.Time.Format.ISO8601 (iso8601Show)
import Data.Type.Equality
import GHC.Conc (listThreads, threadStatus)
import GHC.Conc.Sync (threadLabel)
import GHC.Stats (getRTSStats)
import GHC.TypeLits (KnownNat)
import Network.Socket (ServiceName, Socket, socketToHandle)
import Simplex.Messaging.Agent.Lock
@@ -66,12 +70,12 @@ import Simplex.Messaging.Encoding (Encoding (smpEncode))
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.Control
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.Env.STM as Env
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.MsgStore
import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.STM
import Simplex.Messaging.Server.QueueStore.STM as QS
import Simplex.Messaging.Server.Stats
import Simplex.Messaging.Server.StoreLog
import Simplex.Messaging.TMap (TMap)
@@ -112,8 +116,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
restoreServerMessages
restoreServerStats
raceAny_
( serverThread s subscribedQ subscribers subscriptions cancelSub :
serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
( serverThread s "server subscribedQ" subscribedQ subscribers subscriptions cancelSub :
serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubscriptions (\_ -> pure ()) :
map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
)
`finally` withLock (savingLock s) "final" (saveServer False)
@@ -129,12 +133,14 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
serverThread ::
forall s.
Server ->
String ->
(Server -> TQueue (QueueId, Client)) ->
(Server -> TMap QueueId Client) ->
(Client -> TMap QueueId s) ->
(s -> IO ()) ->
M ()
serverThread s subQ subs clientSubs unsub = forever $ do
serverThread s label subQ subs clientSubs unsub = forever $ do
labelMyThread label
atomically updateSubscribers
$>>= endPreviousSubscriptions
>>= liftIO . mapM_ unsub
@@ -151,6 +157,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
TM.lookupInsert qId clnt (subs s) $>>= clientToBeNotified
endPreviousSubscriptions :: (QueueId, Client) -> M (Maybe s)
endPreviousSubscriptions (qId, c) = do
labelMyThread $ label <> ".endPreviousSubscriptions"
void . forkIO . atomically $
writeTBQueue (sndQ c) [(CorrId "", qId, END)]
atomically $ TM.lookupDelete qId (clientSubs c)
@@ -164,6 +171,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
ms <- asks msgStore
quota <- asks $ msgQueueQuota . config
let interval = checkInterval expCfg * 1000000
labelMyThread "expireMessages"
forever $ do
liftIO $ threadDelay' interval
old <- liftIO $ expireBeforeEpoch expCfg
@@ -179,6 +187,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
logServerStats :: Int64 -> Int64 -> FilePath -> M ()
logServerStats startAt logInterval statsFilePath = do
labelMyThread "logServerStats"
initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
@@ -223,9 +232,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
threadDelay' interval
runClient :: Transport c => TProxy c -> c -> M ()
runClient _ h = do
runClient tp h = do
kh <- asks serverIdentity
smpVRange <- asks $ smpServerVRange . config
labelMyThread $ "smp handshake for " <> transportName tp
liftIO (runExceptT $ smpServerHandshake h kh smpVRange) >>= \case
Right th -> runClientTransport th
Left _ -> pure ()
@@ -239,10 +249,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
srv <- asks server
cpStarted <- newEmptyTMVarIO
u <- askUnliftIO
liftIO $ runTCPServer cpStarted port $ runCPClient u srv
liftIO $ do
labelMyThread "control port server"
runTCPServer cpStarted port $ runCPClient u srv
where
runCPClient :: UnliftIO (ReaderT Env IO) -> Server -> Socket -> IO ()
runCPClient u srv sock = do
labelMyThread "control port client"
h <- socketToHandle sock ReadWriteMode
hSetBuffering h LineBuffering
hSetNewlineMode h universalNewlineMode
@@ -258,7 +271,15 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
processCP h = \case
CPSuspend -> hPutStrLn h "suspend not implemented"
CPResume -> hPutStrLn h "resume not implemented"
CPClients -> hPutStrLn h "clients not implemented"
CPClients -> do
Server {subscribers} <- unliftIO u $ asks server
clients <- readTVarIO subscribers
hPutStrLn h $ "Clients: " <> show (length clients)
forM_ (M.toList clients) $ \(cid, Client {sessionId, connected, activeAt, subscriptions}) -> do
hPutStrLn h . B.unpack $ "Client " <> encode cid <> " $" <> encode sessionId
readTVarIO connected >>= hPutStrLn h . (" connected: " <>) . show
readTVarIO activeAt >>= hPutStrLn h . (" activeAt: " <>) . B.unpack . strEncode
readTVarIO subscriptions >>= hPutStrLn h . (" subscriptions: " <>) . show . M.size
CPStats -> do
ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgSentNtf, msgRecvNtf, qCount, msgCount} <- unliftIO u $ asks serverStats
putStat "fromTime" fromTime
@@ -274,12 +295,21 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
where
putStat :: Show a => String -> TVar a -> IO ()
putStat label var = readTVarIO var >>= \v -> hPutStrLn h $ label <> ": " <> show v
CPStatsRTS -> getRTSStats >>= hPutStrLn h . show
CPThreads -> do
threads <- liftIO listThreads
hPutStrLn h $ "Threads: " <> show (length threads)
forM_ (sort threads) $ \tid -> do
label <- threadLabel tid
status <- threadStatus tid
hPutStrLn h $ show tid <> " (" <> show status <> ") " <> fromMaybe "" label
CPSave -> withLock (savingLock srv) "control" $ do
hPutStrLn h "saving server state..."
unliftIO u $ saveServer True
hPutStrLn h "server state saved!"
CPHelp -> hPutStrLn h "commands: stats, save, help, quit"
CPHelp -> hPutStrLn h "commands: stats, stats-rts, clients, threads, save, help, quit"
CPQuit -> pure ()
CPSkip -> pure ()
runClientTransport :: Transport c => THandle c -> M ()
runClientTransport th@THandle {thVersion, sessionId} = do
@@ -288,6 +318,7 @@ runClientTransport th@THandle {thVersion, sessionId} = do
c <- atomically $ newClient q thVersion sessionId ts
s <- asks server
expCfg <- asks $ inactiveClientExpiration . config
labelMyThread . B.unpack $ "client $" <> encode c.sessionId
raceAny_ ([liftIO $ send th c, client c s, receive th c] <> disconnectThread_ c expCfg)
`finally` clientDisconnected c
where
@@ -318,12 +349,14 @@ cancelSub sub =
_ -> return ()
receive :: Transport c => THandle c -> Client -> M ()
receive th Client {rcvQ, sndQ, activeAt} = forever $ do
ts <- L.toList <$> liftIO (tGet th)
atomically . writeTVar activeAt =<< liftIO getSystemTime
as <- partitionEithers <$> mapM cmdAction ts
write sndQ $ fst as
write rcvQ $ snd as
receive th Client {rcvQ, sndQ, activeAt, sessionId} = do
labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
forever $ do
ts <- L.toList <$> liftIO (tGet th)
atomically . writeTVar activeAt =<< liftIO getSystemTime
as <- partitionEithers <$> mapM cmdAction ts
write sndQ $ fst as
write rcvQ $ snd as
where
cmdAction :: SignedTransmission ErrorType Cmd -> M (Either (Transmission BrokerMsg) (Maybe QueueRec, Transmission Cmd))
cmdAction (sig, signed, (corrId, queueId, cmdOrError)) =
@@ -337,10 +370,12 @@ receive th Client {rcvQ, sndQ, activeAt} = forever $ do
write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
send :: Transport c => THandle c -> Client -> IO ()
send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = forever $ do
ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
atomically . writeTVar activeAt =<< liftIO getSystemTime
send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = do
labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
forever $ do
ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
atomically . writeTVar activeAt =<< liftIO getSystemTime
where
tOrder :: Transmission BrokerMsg -> Int
tOrder (_, _, cmd) = case cmd of
@@ -349,7 +384,8 @@ send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = forever $ do
_ -> 1
disconnectTransport :: Transport c => THandle c -> client -> (client -> TVar SystemTime) -> ExpirationConfig -> IO ()
disconnectTransport THandle {connection} c activeAt expCfg = do
disconnectTransport THandle {connection, sessionId} c activeAt expCfg = do
labelMyThread . B.unpack $ "client $" <> encode sessionId <> " disconnectTransport"
let interval = checkInterval expCfg * 1000000
forever . liftIO $ do
threadDelay' interval
@@ -405,7 +441,8 @@ dummyKeyEd448 :: C.PublicKey 'C.Ed448
dummyKeyEd448 = "MEMwBQYDK2VxAzoA6ibQc9XpkSLtwrf7PLvp81qW/etiumckVFImCMRdftcG/XopbOSaq9qyLhrgJWKOLyNrQPNVvpMA"
client :: forall m. (MonadUnliftIO m, MonadReader Env m) => Client -> Server -> m ()
client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscribedQ, ntfSubscribedQ, notifiers} =
client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscribedQ, ntfSubscribedQ, notifiers} = do
labelMyThread . B.unpack $ "client $" <> encode clnt.sessionId <> " commands"
forever $
atomically (readTBQueue rcvQ)
>>= mapM processCommand
@@ -726,11 +763,11 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
encrypt msgFlags body =
let encBody = EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) (C.cbNonce msgId') body
in RcvMessage msgId' msgTs' msgFlags encBody
msgId' = msgId (msg :: Message)
msgTs' = msgTs (msg :: Message)
msgId' = msg.msgId
msgTs' = msg.msgTs
setDelivered :: Sub -> Message -> STM Bool
setDelivered s msg = tryPutTMVar (delivered s) $ msgId (msg :: Message)
setDelivered s msg = tryPutTMVar (delivered s) $ msg.msgId
getStoreMsgQueue :: T.Text -> RecipientId -> m MsgQueue
getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
@@ -830,7 +867,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
| maybe True (systemSeconds msgTs >=) old_ -> isNothing <$> writeMsg q msg
| otherwise -> pure False
MessageQuota {} -> writeMsg q msg $> False
when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode msg.msgId
updateMsgV1toV3 QueueRec {rcvDhSecret} RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} = do
let nonce = C.cbNonce msgId
msgBody <- liftEither . first (msgErr "v1 message decryption") $ C.maxLenBS =<< C.cbDecrypt rcvDhSecret nonce body
src/Simplex/Messaging/Server/Control.hs
+9
View File
@@ -11,9 +11,12 @@ data ControlProtocol
| CPResume
| CPClients
| CPStats
| CPStatsRTS
| CPThreads
| CPSave
| CPHelp
| CPQuit
| CPSkip
instance StrEncoding ControlProtocol where
strEncode = \case
@@ -21,16 +24,22 @@ instance StrEncoding ControlProtocol where
CPResume -> "resume"
CPClients -> "clients"
CPStats -> "stats"
CPStatsRTS -> "stats-rts"
CPThreads -> "threads"
CPSave -> "save"
CPHelp -> "help"
CPQuit -> "quit"
CPSkip -> ""
strP =
A.takeTill (== ' ') >>= \case
"suspend" -> pure CPSuspend
"resume" -> pure CPResume
"clients" -> pure CPClients
"stats" -> pure CPStats
"stats-rts" -> pure CPStatsRTS
"threads" -> pure CPThreads
"save" -> pure CPSave
"help" -> pure CPHelp
"quit" -> pure CPQuit
"" -> pure CPSkip
_ -> fail "bad ControlProtocol command"
src/Simplex/Messaging/Transport/Client.hs
-1
View File
@@ -21,7 +21,6 @@ module Simplex.Messaging.Transport.Client
where
import Control.Applicative (optional)
import Control.Monad.Except
import Control.Monad.IO.Unlift
import Data.Aeson (FromJSON (..), ToJSON (..))
import qualified Data.Attoparsec.ByteString.Char8 as A
src/Simplex/Messaging/Transport/HTTP2/Client.hs
+1 -1
View File
@@ -9,7 +9,7 @@ module Simplex.Messaging.Transport.HTTP2.Client where
import Control.Concurrent.Async
import Control.Exception (IOException, try)
import qualified Control.Exception as E
import Control.Monad.Except
import Control.Monad
import Data.ByteString.Char8 (ByteString)
import Data.Functor (($>))
import Data.Time (UTCTime, getCurrentTime)
src/Simplex/Messaging/Transport/Server.hs
+3 -3
View File
@@ -17,9 +17,8 @@ module Simplex.Messaging.Transport.Server
where
import Control.Applicative ((<|>))
import Control.Concurrent.STM (stateTVar)
import Control.Logger.Simple
import Control.Monad.Except
import Control.Monad
import Control.Monad.IO.Unlift
import qualified Crypto.Store.X509 as SX
import Data.Default (def)
@@ -33,7 +32,7 @@ import qualified Network.TLS as T
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Transport
import Simplex.Messaging.Util (catchAll_, tshow)
import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
import System.Exit (exitFailure)
import System.Mem.Weak (Weak, deRefWeak)
import UnliftIO.Concurrent
@@ -64,6 +63,7 @@ runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -
runTransportServer started port serverParams cfg server = do
u <- askUnliftIO
let tCfg = serverTransportConfig cfg
labelMyThread $ "transport server for " <> transportName (TProxy :: TProxy c)
liftIO . runTCPServer started port $ \conn ->
E.bracket
(connectTLS Nothing tCfg serverParams conn >>= getServerConnection tCfg)
src/Simplex/Messaging/Util.hs
+7 -9
View File
@@ -5,10 +5,11 @@ module Simplex.Messaging.Util where
import Control.Concurrent (threadDelay)
import qualified Control.Exception as E
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Unlift
import Control.Monad.Trans.Except
import Data.Bifunctor (first)
import qualified Data.ByteString as BW
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.Int (Int64)
@@ -19,6 +20,8 @@ import Data.Text (Text)
import qualified Data.Text as T
import Data.Text.Encoding (decodeUtf8With)
import Data.Time (NominalDiffTime)
import GHC.Conc
import Numeric (showHex)
import UnliftIO.Async
import qualified UnliftIO.Exception as UE
@@ -66,14 +69,6 @@ liftEitherWith :: (MonadError e' m) => (e -> e') -> Either e a -> m a
liftEitherWith f = liftEither . first f
{-# INLINE liftEitherWith #-}
tryError :: MonadError e m => m a -> m (Either e a)
tryError action = (Right <$> action) `catchError` (pure . Left)
{-# INLINE tryError #-}
tryE :: Monad m => ExceptT e m a -> ExceptT e m (Either e a)
tryE m = (Right <$> m) `catchE` (pure . Left)
{-# INLINE tryE #-}
liftE :: (e -> e') -> ExceptT e IO a -> ExceptT e' IO a
liftE f a = ExceptT $ first f <$> runExceptT a
{-# INLINE liftE #-}
@@ -157,3 +152,6 @@ diffToMicroseconds diff = fromIntegral ((truncate $ diff * 1000000) :: Integer)
diffToMilliseconds :: NominalDiffTime -> Int64
diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)
labelMyThread :: MonadIO m => String -> m ()
labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label)
stack.yaml
+6 -4
View File
@@ -17,7 +17,7 @@
#
# resolver: ./custom-snapshot.yaml
# resolver: https://example.com/snapshots/2018-01-01.yaml
resolver: lts-18.28
resolver: nightly-2023-08-22
# User packages to be built.
# Various formats can be used as shown in the example below.
@@ -47,15 +47,17 @@ extra-deps:
- text-short-0.1.5@sha256:962c6228555debdc46f758d0317dea16e5240d01419b42966674b08a5c3d8fa6,3498
- time-compat-1.9.6.1@sha256:42d8f2e08e965e1718917d54ad69e1d06bd4b87d66c41dc7410f59313dba4ed1,5033
- github: simplex-chat/aeson
commit: 3eb66f9a68f103b5f1489382aad89f5712a64db7
commit: 68330dce8208173c6acf5f62b23acb500ab5d873
- github: kazu-yamamoto/http2
commit: b5a1b7200cf5bc7044af34ba325284271f6dff25
# - ../direct-sqlcipher
- github: simplex-chat/direct-sqlcipher
commit: 34309410eb2069b029b8fc1872deb1e0db123294
commit: f814ee68b16a9447fbb467ccc8f29bdd3546bfd9
# - ../sqlcipher-simple
- github: simplex-chat/sqlcipher-simple
commit: 5e154a2aeccc33ead6c243ec07195ab673137221
commit: a46bd361a19376c5211f1058908fc0ae6bf42446
- github: simplex-chat/network-transport
commit: 0013798272a683e35ca38d2fdaf480942311fba8
# - ../hs-tls/core
# - github: simplex-chat/hs-tls
# commit: f6cc753611f80af300401cfae63846e9d7c40d9e
tests/AgentTests/ConnectionRequestTests.hs
+2
View File
@@ -3,6 +3,8 @@
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module AgentTests.ConnectionRequestTests where
import Data.ByteString (ByteString)
tests/AgentTests/FunctionalAPITests.hs
-1
View File
@@ -60,7 +60,6 @@ import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Transport (ATransport (..))
import Simplex.Messaging.Util (tryError)
import Simplex.Messaging.Version
import System.Directory (copyFile, renameFile)
import Test.Hspec
tests/AgentTests/NotificationTests.hs
+2 -1
View File
@@ -11,7 +11,9 @@ module AgentTests.NotificationTests where
-- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
import AgentTests.FunctionalAPITests (exchangeGreetingsMsgId, get, getSMPAgentClient', makeConnection, nGet, runRight, runRight_, switchComplete, testServerMatrix2, (##>), (=##>), pattern Msg)
import Control.Concurrent (killThread, threadDelay)
import Control.Monad
import Control.Monad.Except
import Control.Monad.Trans.Except
import qualified Data.Aeson as J
import qualified Data.Aeson.Types as JT
import Data.Bifunctor (bimap, first)
@@ -33,7 +35,6 @@ import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsg
import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
import Simplex.Messaging.Transport (ATransport)
import Simplex.Messaging.Util (tryE)
import System.Directory (doesFileExist, removeFile)
import Test.Hspec
import UnliftIO
tests/AgentTests/SQLiteTests.hs
+2
View File
@@ -7,6 +7,8 @@
{-# LANGUAGE QuasiQuotes #-}
{-# LANGUAGE RecordWildCards #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module AgentTests.SQLiteTests (storeTests) where
import Control.Concurrent.Async (concurrently_)
tests/CoreTests/BatchingTests.hs
+1 -1
View File
@@ -3,7 +3,7 @@
module CoreTests.BatchingTests (batchingTests) where
import Control.Concurrent.STM
import Control.Monad.Except
import Control.Monad
import Crypto.Random (MonadRandom(..))
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
tests/CoreTests/CryptoFileTests.hs
+1
View File
@@ -4,6 +4,7 @@ module CoreTests.CryptoFileTests (cryptoFileTests) where
import AgentTests.FunctionalAPITests (runRight_)
import Control.Monad.Except
import Control.Monad.IO.Class
import Crypto.Random (getRandomBytes)
import qualified Data.ByteString.Lazy as LB
import GHC.IO.IOMode (IOMode (..))
tests/CoreTests/UtilTests.hs
+1
View File
@@ -4,6 +4,7 @@ module CoreTests.UtilTests where
import Control.Exception (Exception, SomeException, throwIO)
import Control.Monad.Except
import Control.Monad.IO.Class
import Data.IORef
import Simplex.Messaging.Util
import Simplex.Messaging.Client.Agent ()
tests/NtfClient.hs
+3 -2
View File
@@ -5,8 +5,8 @@
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StandaloneDeriving #-}
@@ -133,8 +133,9 @@ ntfServerTest ::
IO (Maybe C.ASignature, ByteString, ByteString, BrokerMsg)
ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
where
tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
tPut' h (sig, corrId, queueId, smp) = do
let t' = smpEncode (sessionId (h :: THandle c), corrId, queueId, smp)
let t' = smpEncode (h.sessionId, corrId, queueId, smp)
[Right ()] <- tPut h Nothing [(sig, t')]
pure ()
tGet' h = do
tests/SMPClient.hs
+3 -1
View File
@@ -4,6 +4,7 @@
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
@@ -159,8 +160,9 @@ smpServerTest ::
IO (Maybe C.ASignature, ByteString, ByteString, BrokerMsg)
smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
where
tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
tPut' h (sig, corrId, queueId, smp) = do
let t' = smpEncode (sessionId (h :: THandle c), corrId, queueId, smp)
let t' = smpEncode (h.sessionId, corrId, queueId, smp)
[Right ()] <- tPut h Nothing [(sig, t')]
pure ()
tGet' h = do
tests/ServerTests.hs
+1 -1
View File
@@ -15,7 +15,7 @@ import AgentTests.NotificationTests (removeFileIfExists)
import Control.Concurrent (ThreadId, killThread, threadDelay)
import Control.Concurrent.STM
import Control.Exception (SomeException, try)
import Control.Monad.Except (forM, forM_)
import Control.Monad
import Control.Monad.IO.Class
import Data.Bifunctor (first)
import Data.ByteString.Base64
tests/XFTPAgent.hs
+2
View File
@@ -9,7 +9,9 @@ module XFTPAgent where
import AgentTests.FunctionalAPITests (get, getSMPAgentClient', rfGet, runRight, runRight_, sfGet)
import Control.Concurrent (threadDelay)
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Class
import qualified Data.ByteString.Char8 as B
import qualified Data.ByteString.Lazy as LB
import Data.Int (Int64)
tests/XFTPServerTests.hs
+3 -1
View File
@@ -10,7 +10,9 @@ import AgentTests.FunctionalAPITests (runRight_)
import Control.Concurrent (threadDelay)
import Control.Concurrent.STM
import Control.Exception (SomeException)
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Unlift
import Crypto.Random (getRandomBytes)
import qualified Data.ByteString.Base64.URL as B64
import Data.ByteString.Char8 (ByteString)
@@ -140,7 +142,7 @@ runTestFileChunkDelete s r = do
deleteXFTPChunk s spKey sId
liftIO $
readChunk sId
`shouldThrow` \(e :: SomeException) -> "openBinaryFile: does not exist" `isInfixOf` show e
`shouldThrow` \(e :: SomeException) -> "withBinaryFile: does not exist" `isInfixOf` show e
downloadXFTPChunk r rpKey rId (XFTPRcvChunkSpec "tests/tmp/received_chunk2" chSize digest)
`catchError` (liftIO . (`shouldBe` PCEProtocolError AUTH))
deleteXFTPChunk s spKey sId