Skip to content

Commit b61912a

Browse files
committed
Keep exception in case of error
... also rename because we can't assume these are always auth errors
1 parent 2e7b233 commit b61912a

2 files changed

Lines changed: 171 additions & 157 deletions

File tree

‎core/src/main/scala/fs2/io/ssh/Client.scala‎

Lines changed: 47 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -41,57 +41,54 @@ final class Client[F[_]] private (client: SshClient)(implicit F: Async[F]) {
4141
import Client.Error
4242
import MinaFuture._
4343

44-
@SuppressWarnings(
45-
Array(
46-
"org.wartremover.warts.DefaultArguments"))
47-
def exec(
48-
cc: ConnectionConfig,
49-
command: String,
50-
chunkSize: Int = 4096)(
51-
implicit FR: Raise[F, Error])
52-
: Resource[F, Process[F]] = {
44+
@SuppressWarnings(Array("org.wartremover.warts.DefaultArguments"))
45+
def exec(cc: ConnectionConfig, command: String, chunkSize: Int = 4096)(
46+
implicit FR: Raise[F, Error]
47+
): Resource[F, Process[F]] = {
5348

5449
for {
5550
session <- this.session(cc)
5651

57-
channel <- Resource.make(
58-
for {
59-
channel <- F.delay(session.createExecChannel(command))
60-
_ <- F.delay(channel.setStreaming(StreamingChannel.Streaming.Async))
61-
opened <- fromFuture(F.delay(channel.open()))
62-
// TODO handle failure opening
63-
} yield channel)(
64-
channel => fromFuture(F.delay(channel.close(false))).void)
52+
channel <- Resource.make(for {
53+
channel <- F.delay(session.createExecChannel(command))
54+
_ <- F.delay(channel.setStreaming(StreamingChannel.Streaming.Async))
55+
opened <- fromFuture(F.delay(channel.open()))
56+
// TODO handle failure opening
57+
} yield channel)(channel =>
58+
fromFuture(F.delay(channel.close(false))).void
59+
)
6560
} yield new Process[F](channel, chunkSize)
6661
}
6762

6863
def portForward(
69-
cc: ConnectionConfig,
70-
local: InetSocketAddress,
71-
remote: InetSocketAddress)(
72-
implicit FR: Raise[F, Error])
73-
: Resource[F, Unit] = {
74-
this.session(cc).flatMap { clientSession =>
75-
val l = SshdSocketAddress.toSshdSocketAddress(local)
76-
val r = SshdSocketAddress.toSshdSocketAddress(remote)
77-
Resource.make(F.delay(clientSession.startLocalPortForwarding(l, r)))(address => F.delay(clientSession.stopLocalPortForwarding(address)))
78-
}.void
64+
cc: ConnectionConfig,
65+
local: InetSocketAddress,
66+
remote: InetSocketAddress
67+
)(implicit FR: Raise[F, Error]): Resource[F, Unit] = {
68+
this
69+
.session(cc)
70+
.flatMap { clientSession =>
71+
val l = SshdSocketAddress.toSshdSocketAddress(local)
72+
val r = SshdSocketAddress.toSshdSocketAddress(remote)
73+
Resource.make(F.delay(clientSession.startLocalPortForwarding(l, r)))(
74+
address => F.delay(clientSession.stopLocalPortForwarding(address))
75+
)
76+
}
77+
.void
7978
}
8079

8180
@SuppressWarnings(
82-
Array(
83-
"org.wartremover.warts.Null",
84-
"org.wartremover.warts.ToString"))
81+
Array("org.wartremover.warts.Null", "org.wartremover.warts.ToString")
82+
)
8583
def session(
86-
cc: ConnectionConfig)(
87-
implicit FR: Raise[F, Error])
88-
: Resource[F, ClientSession] = {
84+
cc: ConnectionConfig
85+
)(implicit FR: Raise[F, Error]): Resource[F, ClientSession] = {
8986

9087
for {
9188
session <-
9289
Resource.make(
93-
fromFuture(F.delay(client.connect(cc.username, cc.host))))(
94-
session => fromFuture(F.delay(session.close(false))).void)
90+
fromFuture(F.delay(client.connect(cc.username, cc.host)))
91+
)(session => fromFuture(F.delay(session.close(false))).void)
9592

9693
_ <- cc.auth match {
9794
case Auth.Password(text) =>
@@ -115,7 +112,9 @@ final class Client[F[_]] private (client: SshClient)(implicit F: Async[F]) {
115112
}
116113

117114
case None =>
118-
F.delay(provider.setPasswordFinder(FilePasswordProvider.EMPTY))
115+
F.delay(
116+
provider.setPasswordFinder(FilePasswordProvider.EMPTY)
117+
)
119118
}
120119

121120
pairs <- F.blocking(provider.loadKeys(session))
@@ -142,8 +141,8 @@ final class Client[F[_]] private (client: SshClient)(implicit F: Async[F]) {
142141

143142
_ <- Resource eval {
144143
success match {
145-
case Left(_: SshException) =>
146-
FR.raise(Error.Authentication)
144+
case Left(ex: SshException) =>
145+
FR.raise(Error.SshErr(ex))
147146

148147
case Left(e) =>
149148
F.raiseError(e)
@@ -168,7 +167,11 @@ object Client {
168167

169168
// send periodic SSH_MSG_IGNOREs to prevent the connection from dying
170169
// without activity
171-
client.setSessionHeartbeat(HeartbeatType.IGNORE, TimeUnit.SECONDS, HeartbeatInterval)
170+
client.setSessionHeartbeat(
171+
HeartbeatType.IGNORE,
172+
TimeUnit.SECONDS,
173+
HeartbeatInterval
174+
)
172175

173176
client.start()
174177
client
@@ -178,15 +181,14 @@ object Client {
178181
}
179182

180183
// convenience function that really should live elsewhere
181-
def resolve[F[_]: Async](
182-
hostname: String,
183-
port: Int)
184-
: F[InetSocketAddress] =
185-
Async[F].blocking(new InetSocketAddress(InetAddress.getByName(hostname), port))
184+
def resolve[F[_]: Async](hostname: String, port: Int): F[InetSocketAddress] =
185+
Async[F].blocking(
186+
new InetSocketAddress(InetAddress.getByName(hostname), port)
187+
)
186188

187189
sealed trait Error extends Product with Serializable
188190

189191
object Error {
190-
case object Authentication extends Error
192+
case class SshErr(ex: SshException) extends Error
191193
}
192194
}

0 commit comments

Comments
 (0)