nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Reduces the amount of competing threads to restart connections with relays.

Browse Source
This commit is contained in:
Vitor Pamplona
2023-10-20 12:37:05 -04:00
parent 2892fa6fca
commit 1ab93adc22
6 changed files with 39 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/main/java/com/vitorpamplona/amethyst/ServiceManager.kt
+9 -1
View File
@@ -43,7 +43,7 @@ object ServiceManager {
private var isStarted: Boolean = false // to not open amber in a loop trying to use auth relays and registering for notifications private var isStarted: Boolean = false // to not open amber in a loop trying to use auth relays and registering for notifications
private var account: Account? = null private var account: Account? = null
fun start(account: Account) { private fun start(account: Account) {
this.account = account this.account = account
ExternalSignerUtils.account = account ExternalSignerUtils.account = account
start() start()
@@ -51,6 +51,7 @@ object ServiceManager {
@Synchronized @Synchronized
fun start() { fun start() {
Log.d("ServiceManager", "Pre Starting Relay Services $isStarted $account")
if (isStarted && account != null) { if (isStarted && account != null) {
return return
} }
@@ -158,6 +159,13 @@ object ServiceManager {
start(account) start(account)
} }
} }
fun forceRestartIfItShould() {
if (shouldPauseService) {
pause()
start()
}
}
} }
object SingletonDiskCache { object SingletonDiskCache {
app/src/main/java/com/vitorpamplona/amethyst/service/NostrDataSource.kt
+2 -2
View File
@@ -50,7 +50,7 @@ abstract class NostrDataSource(val debugName: String) {
// Log.e("ERROR", "Relay ${relay.url}: ${error.message}") // Log.e("ERROR", "Relay ${relay.url}: ${error.message}")
} }
override fun onRelayStateChange(type: Relay.Type, relay: Relay, subscriptionId: String?) { override fun onRelayStateChange(type: Relay.StateType, relay: Relay, subscriptionId: String?) {
// Log.d("RELAY", "Relay ${relay.url} ${when (type) { // Log.d("RELAY", "Relay ${relay.url} ${when (type) {
// Relay.Type.CONNECT -> "connected." // Relay.Type.CONNECT -> "connected."
// Relay.Type.DISCONNECT -> "disconnected." // Relay.Type.DISCONNECT -> "disconnected."
@@ -58,7 +58,7 @@ abstract class NostrDataSource(val debugName: String) {
// Relay.Type.EOSE -> "sent all events it had stored." // Relay.Type.EOSE -> "sent all events it had stored."
// }}") // }}")
if (type == Relay.Type.EOSE && subscriptionId != null && subscriptionId in subscriptions.keys) { if (type == Relay.StateType.EOSE && subscriptionId != null && subscriptionId in subscriptions.keys) {
// updates a per subscripton since date // updates a per subscripton since date
subscriptions[subscriptionId]?.updateEOSE( subscriptions[subscriptionId]?.updateEOSE(
TimeUtils.fiveMinutesAgo(), // in case people's clock is slighly off. TimeUtils.fiveMinutesAgo(), // in case people's clock is slighly off.
app/src/main/java/com/vitorpamplona/amethyst/service/relays/Client.kt
+2 -2
View File
@@ -151,7 +151,7 @@ object Client : RelayPool.Listener {
} }
@OptIn(DelicateCoroutinesApi::class) @OptIn(DelicateCoroutinesApi::class)
override fun onRelayStateChange(type: Relay.Type, relay: Relay, channel: String?) { override fun onRelayStateChange(type: Relay.StateType, relay: Relay, channel: String?) {
// Releases the Web thread for the new payload. // Releases the Web thread for the new payload.
// May need to add a processing queue if processing new events become too costly. // May need to add a processing queue if processing new events become too costly.
GlobalScope.launch(Dispatchers.Default) { GlobalScope.launch(Dispatchers.Default) {
@@ -207,7 +207,7 @@ object Client : RelayPool.Listener {
/** /**
* Connected to or disconnected from a relay * Connected to or disconnected from a relay
*/ */
open fun onRelayStateChange(type: Relay.Type, relay: Relay, channel: String?) = Unit open fun onRelayStateChange(type: Relay.StateType, relay: Relay, channel: String?) = Unit
/** /**
* When an relay saves or rejects a new event. * When an relay saves or rejects a new event.
app/src/main/java/com/vitorpamplona/amethyst/service/relays/Relay.kt
+9 -8
View File
@@ -128,7 +128,7 @@ class Relay(
// Log.w("Relay", "Relay OnOpen, Loading All subscriptions $url") // Log.w("Relay", "Relay OnOpen, Loading All subscriptions $url")
onConnected(this@Relay) onConnected(this@Relay)
listeners.forEach { it.onRelayStateChange(this@Relay, Type.CONNECT, null) } listeners.forEach { it.onRelayStateChange(this@Relay, StateType.CONNECT, null) }
} }
override fun onMessage(webSocket: WebSocket, text: String) { override fun onMessage(webSocket: WebSocket, text: String) {
@@ -152,7 +152,7 @@ class Relay(
listeners.forEach { listeners.forEach {
it.onRelayStateChange( it.onRelayStateChange(
this@Relay, this@Relay,
Type.DISCONNECTING, StateType.DISCONNECTING,
null null
) )
} }
@@ -163,7 +163,7 @@ class Relay(
markConnectionAsClosed() markConnectionAsClosed()
listeners.forEach { it.onRelayStateChange(this@Relay, Type.DISCONNECT, null) } listeners.forEach { it.onRelayStateChange(this@Relay, StateType.DISCONNECT, null) }
} }
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) { override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
@@ -211,14 +211,14 @@ class Relay(
listeners.forEach { listeners.forEach {
it.onEvent(this@Relay, channel, event) it.onEvent(this@Relay, channel, event)
if (afterEOSE) { if (afterEOSE) {
it.onRelayStateChange(this@Relay, Type.EOSE, channel) it.onRelayStateChange(this@Relay, StateType.EOSE, channel)
} }
} }
} }
"EOSE" -> listeners.forEach { "EOSE" -> listeners.forEach {
afterEOSE = true afterEOSE = true
// Log.w("Relay", "Relay onEOSE $url, $channel") // Log.w("Relay", "Relay onEOSE $url, $channel")
it.onRelayStateChange(this@Relay, Type.EOSE, channel) it.onRelayStateChange(this@Relay, StateType.EOSE, channel)
} }
"NOTICE" -> listeners.forEach { "NOTICE" -> listeners.forEach {
Log.w("Relay", "Relay onNotice $url, $channel") Log.w("Relay", "Relay onNotice $url, $channel")
@@ -256,7 +256,8 @@ class Relay(
} }
fun disconnect() { fun disconnect() {
// httpClient.dispatcher.executorService.shutdown() checkNotInMainThread()
closingTimeInSeconds = TimeUtils.now() closingTimeInSeconds = TimeUtils.now()
socket?.close(1000, "Normal close") socket?.close(1000, "Normal close")
socket = null socket = null
@@ -360,7 +361,7 @@ class Relay(
activeTypes == other.activeTypes activeTypes == other.activeTypes
} }
enum class Type { enum class StateType {
// Websocket connected // Websocket connected
CONNECT, CONNECT,
@@ -391,6 +392,6 @@ class Relay(
* *
* @param type is 0 for disconnect and 1 for connect * @param type is 0 for disconnect and 1 for connect
*/ */
fun onRelayStateChange(relay: Relay, type: Type, channel: String?) fun onRelayStateChange(relay: Relay, type: StateType, channel: String?)
} }
} }
app/src/main/java/com/vitorpamplona/amethyst/service/relays/RelayPool.kt
+3 -3
View File
@@ -107,7 +107,7 @@ object RelayPool : Relay.Listener {
fun onError(error: Error, subscriptionId: String, relay: Relay) fun onError(error: Error, subscriptionId: String, relay: Relay)
fun onRelayStateChange(type: Relay.Type, relay: Relay, channel: String?) fun onRelayStateChange(type: Relay.StateType, relay: Relay, channel: String?)
fun onSendResponse(eventId: String, success: Boolean, message: String, relay: Relay) fun onSendResponse(eventId: String, success: Boolean, message: String, relay: Relay)
@@ -123,9 +123,9 @@ object RelayPool : Relay.Listener {
updateStatus() updateStatus()
} }
override fun onRelayStateChange(relay: Relay, type: Relay.Type, channel: String?) { override fun onRelayStateChange(relay: Relay, type: Relay.StateType, channel: String?) {
listeners.forEach { it.onRelayStateChange(type, relay, channel) } listeners.forEach { it.onRelayStateChange(type, relay, channel) }
if (type != Relay.Type.EOSE) { if (type != Relay.StateType.EOSE) {
updateStatus() updateStatus()
} }
} }
app/src/main/java/com/vitorpamplona/amethyst/ui/MainActivity.kt
+14 -38
View File
@@ -6,7 +6,6 @@ import android.content.Intent
import android.net.ConnectivityManager import android.net.ConnectivityManager
import android.net.Network import android.net.Network
import android.net.NetworkCapabilities import android.net.NetworkCapabilities
import android.net.NetworkRequest
import android.os.Build import android.os.Build
import android.os.Bundle import android.os.Bundle
import android.util.Log import android.util.Log
@@ -91,8 +90,8 @@ class MainActivity : AppCompatActivity() {
DefaultMutedSetting.value = true DefaultMutedSetting.value = true
// Only starts after login // Only starts after login
GlobalScope.launch(Dispatchers.IO) { if (ServiceManager.shouldPauseService) {
if (ServiceManager.shouldPauseService) { GlobalScope.launch(Dispatchers.IO) {
ServiceManager.start() ServiceManager.start()
} }
} }
@@ -101,24 +100,21 @@ class MainActivity : AppCompatActivity() {
PushNotificationUtils.init(LocalPreferences.allSavedAccounts()) PushNotificationUtils.init(LocalPreferences.allSavedAccounts())
} }
val networkRequest = NetworkRequest.Builder() (getSystemService(ConnectivityManager::class.java) as ConnectivityManager).registerDefaultNetworkCallback(networkCallback)
.addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
.addTransportType(NetworkCapabilities.TRANSPORT_WIFI)
.addTransportType(NetworkCapabilities.TRANSPORT_CELLULAR)
.build()
(getSystemService(ConnectivityManager::class.java) as ConnectivityManager)
.registerNetworkCallback(networkRequest, networkCallback)
} }
override fun onPause() { override fun onPause() {
ServiceManager.cleanObservers() ServiceManager.cleanObservers()
// if (BuildConfig.DEBUG) { // if (BuildConfig.DEBUG) {
debugState(this) GlobalScope.launch(Dispatchers.IO) {
debugState(this@MainActivity)
}
// } // }
if (ServiceManager.shouldPauseService) { if (ServiceManager.shouldPauseService) {
ServiceManager.pause() GlobalScope.launch(Dispatchers.IO) {
ServiceManager.pause()
}
} }
(getSystemService(ConnectivityManager::class.java) as ConnectivityManager) (getSystemService(ConnectivityManager::class.java) as ConnectivityManager)
@@ -149,19 +145,6 @@ class MainActivity : AppCompatActivity() {
@OptIn(DelicateCoroutinesApi::class) @OptIn(DelicateCoroutinesApi::class)
private val networkCallback = object : ConnectivityManager.NetworkCallback() { private val networkCallback = object : ConnectivityManager.NetworkCallback() {
// network is available for use
override fun onAvailable(network: Network) {
super.onAvailable(network)
Log.d("NETWORKCALLBACK", "onAvailable: Disconnecting and connecting again")
// Only starts after login
GlobalScope.launch(Dispatchers.IO) {
if (ServiceManager.shouldPauseService) {
ServiceManager.pause()
ServiceManager.start()
}
}
}
// Network capabilities have changed for the network // Network capabilities have changed for the network
override fun onCapabilitiesChanged( override fun onCapabilitiesChanged(
network: Network, network: Network,
@@ -171,20 +154,13 @@ class MainActivity : AppCompatActivity() {
GlobalScope.launch(Dispatchers.IO) { GlobalScope.launch(Dispatchers.IO) {
val isOnMobileData = networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) val isOnMobileData = networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR)
Log.d("NETWORKCALLBACK", "onCapabilitiesChanged: hasMobileData $isOnMobileData") val isOnWifi = networkCapabilities.hasTransport(NetworkCapabilities.TRANSPORT_WIFI)
Log.d("ServiceManager NetworkCallback", "onCapabilitiesChanged: ${network.networkHandle} hasMobileData $isOnMobileData hasWifi $isOnWifi")
isOnMobileDataState.value = isOnMobileData if (isOnMobileDataState.value != isOnMobileData) {
} isOnMobileDataState.value = isOnMobileData
}
// lost network connection ServiceManager.forceRestartIfItShould()
override fun onLost(network: Network) {
super.onLost(network)
Log.d("NETWORKCALLBACK", "onLost: Disconnecting and pausing relay's connection")
// Only starts after login
GlobalScope.launch(Dispatchers.IO) {
if (ServiceManager.shouldPauseService) {
ServiceManager.pause()
} }
} }
} }