[akka-user] Re: TCP Handshaking custom stage drives me crazy

[Henrik Larsson]

Thanks for the help, yes that did the trick to add one more pull for sslIn. 
The reason bytesIn is empty is because I really dont need a bidishape. I 
need this logic encapsulated in something with one input and two outputs 
because after this handshaking is done there is no need to send any more 
messages to the TCP socket. However since I found your example the only 
example that we close to what i needed and you use bidishape i stick with 
this solution until i understad more about Akka Stream.

On Friday, June 2, 2017 at 1:40:42 PM UTC+2, Michał Sitko wrote:
>
> Hey Larsson - based on debug messages when running live I would guess that 
> you miss `pull(sslIn)` somewhere. I have a hypothesis for what's going on. 
> Let's take a look at:
>
> ```
> setHandler(bytesOut, new OutHandler {
>   override def onPull() = state match {
> case Connecting =>
> case Authenticating =>
> case Subscribing =>
> case Subscribed => pull(sslIn)
>   }
> })
> ```
>
> You're pulling just when in `Subscribed` state. That means that if 
> `bytesOut.onPull` was called before Stage went into `Subscribed` state then 
> `pull(sslIn)` will not get called. Therefore we need to ensure that pull 
> will be called even in that case. You can do this by e.g. adding:
>
> ```scala
> if (subscriptionValid(event)) {
>   state = Subscribed
>   logger.info(state.toString)
>   if (isAvailable(sslIn)) {
> pull(sslIn)
>   }
> }
> ```
>
> in `onPush` handler for `sslIn`.
>
> To debug it I would add `println`s in all places we do `pull(sslIn)` just 
> to be sure they're really are executed. In extreme case you can add println 
> before all calls to `pull` and `push` - I know there will be a lot of 
> output but will give you insight into what's going on.
>
> BTW, I don't understand this one:
>
> ```scala
> setHandler(bytesIn, new InHandler {
>   override def onPush() = {
>   }
> }
> ```
>
> Also, find it strange that your test works. You are not sending anything 
> with `toBetfairProbe` (which simulates input port from TCP as far as I 
> understand). I have not idea, maybe you pasted wrong code? 
>
> Hope it will help, let us know in case of further troubles.
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Re: TCP Handshaking custom stage drives me crazy

[Henrik Larsson]

This is the output generated by the test wich is as expected.
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
ConnectionMessage(connection,050-250517141605-1391626)
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Authenticating
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
StatusMessage(status,Some(1),None,None,None,false,SUCCESS)
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Subscribing
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
StatusMessage(status,Some(123),None,None,None,false,SUCCESS)
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Subscribed
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
MarketChangeMessage(mcm,123,Some(HEARTBEAT),,None,1495773941868,None,None,None,None)
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Heartbeat
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
MarketChangeMessage(mcm,123,Some(BOOM),,None,1495773941868,None,None,None,None)
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - MarketChange
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
MarketChangeMessage(mcm,123,Some(HEARTBEAT),,None,1495773941868,None,None,None,None)
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Heartbeat
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
MarketChangeMessage(mcm,123,Some(BOOM),,None,1495773941868,None,None,None,None)
20:45:53 [default-akka.actor.default-dispatcher-2] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - MarketChange

However when running this live this is the output:
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
ConnectionMessage(connection,050-250517141605-1391626)
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Authenticating
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
StatusMessage(status,Some(1),None,None,None,false,SUCCESS)
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Subscribing
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - 
StatusMessage(status,Some(123),None,None,None,false,SUCCESS)
20:45:53 [default-akka.actor.default-dispatcher-4] INFO 
 c.b.i.b.temp.SubscriptionGraphStage - Subscribed

On Tuesday, May 30, 2017 at 8:41:44 PM UTC+2, Henrik Larsson wrote:
>
>
> *Im trying to construct a GaphStage that will be used when connecting to a 
> TCP socket. The following protocol is used when connecting:
> 1. Materialize TCP flow to create an outgoingConnection
> 2. Socket return Connect message
> 3. If Connect message OK send AuthenticationMessage
> 4. If OK response on AuthenticationMessage send SubscriptionMessage
> 5. If Subscription OK forward all incomming messages that are not Heatbeat
>
> This is the code for the GraphStage.*
>
>
>
> class SubscriptionGraphStage(token: SessionToken)
>   extends GraphStage[BidiShape[BetfairRequest, BetfairRequest, BetfairEvent, 
> BetfairEvent]]
> with LazyLogging {
>
>   import com.bfg.infrastructure.betfair.temp.TcpProxyState._
>
>   val bytesIn: Inlet[BetfairRequest] = Inlet("OutgoingTCP.in")
>
>   val bytesOut: Outlet[BetfairEvent] = Outlet("OutgoingTCP.out")
>   val sslIn: Inlet[BetfairEvent] = Inlet("OutgoingSSL.in")
>   val sslOut: Outlet[BetfairRequest] = Outlet("OutgoingSSL.out")
>
>   override def shape: BidiShape[BetfairRequest, BetfairRequest, BetfairEvent, 
> BetfairEvent] = BidiShape.apply(bytesIn, sslOut, sslIn, bytesOut)
>
>   val authMessage = RequestAuthentication(session = token.sessionToken, id = 
> 1, appKey = "tDwhr80fJKsOW725")
>   val marketSubscription: BetfairRequest = RequestMarketSubscription(
> id=123,
> marketFilter = MarketSubscriptionMarketFilter(
>   eventTypeIds = List("7"),
>   marketTypes = List("WIN"),
>   countryCodes = List("GB")
> ),
> marketDataFilter = MarketSubscriptionMarketDataFilter(
>   fields = List("EX_ALL_OFFERS","EX_MARKET_DEF"),
>   ladderLevels = Some(3)
> )
>   )
>   def connectionValid(event: BetfairEvent): Boolean = event match {
> case m: ConnectionMessage => true
> case _ => false
&

[akka-user] TCP Handshaking custom stage drives me crazy

[Henrik Larsson]

*Im trying to construct a GaphStage that will be used when connecting to a TCP 
socket. The following protocol is used when connecting:
1. Materialize TCP flow to create an outgoingConnection
2. Socket return Connect message
3. If Connect message OK send AuthenticationMessage
4. If OK response on AuthenticationMessage send SubscriptionMessage
5. If Subscription OK forward all incomming messages that are not Heatbeat

This is the code for the GraphStage.*



class SubscriptionGraphStage(token: SessionToken)
  extends GraphStage[BidiShape[BetfairRequest, BetfairRequest, BetfairEvent, 
BetfairEvent]]
with LazyLogging {

  import com.bfg.infrastructure.betfair.temp.TcpProxyState._

  val bytesIn: Inlet[BetfairRequest] = Inlet("OutgoingTCP.in")

  val bytesOut: Outlet[BetfairEvent] = Outlet("OutgoingTCP.out")
  val sslIn: Inlet[BetfairEvent] = Inlet("OutgoingSSL.in")
  val sslOut: Outlet[BetfairRequest] = Outlet("OutgoingSSL.out")

  override def shape: BidiShape[BetfairRequest, BetfairRequest, BetfairEvent, 
BetfairEvent] = BidiShape.apply(bytesIn, sslOut, sslIn, bytesOut)

  val authMessage = RequestAuthentication(session = token.sessionToken, id = 1, 
appKey = "tDwhr80fJKsOW725")
  val marketSubscription: BetfairRequest = RequestMarketSubscription(
id=123,
marketFilter = MarketSubscriptionMarketFilter(
  eventTypeIds = List("7"),
  marketTypes = List("WIN"),
  countryCodes = List("GB")
),
marketDataFilter = MarketSubscriptionMarketDataFilter(
  fields = List("EX_ALL_OFFERS","EX_MARKET_DEF"),
  ladderLevels = Some(3)
)
  )
  def connectionValid(event: BetfairEvent): Boolean = event match {
case m: ConnectionMessage => true
case _ => false
  }

  def authenticationValid(event: BetfairEvent): Boolean = event match {
//todo simplifed need to control id
case m: StatusMessage if m.statusCode == "SUCCESS" => true
case _ => false
  }
  def subscriptionValid(event: BetfairEvent): Boolean = event match {
//todo simplifed need to control id
case m: StatusMessage if m.statusCode == "SUCCESS" => true
case _ => false
  }

  def heatbeatValid(event: BetfairEvent): Boolean = event match {
case m: MarketChangeMessage if m.ct == Some("HEARTBEAT") => true
case _ => false
  }

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = 
new GraphStageLogic(shape) {
private var state: State = Connecting

// I1
setHandler(sslIn, new InHandler {
  override def onPush() = {
state match {
  case Connecting =>
val event = grab(sslIn)
logger.info(event.toString)
if (connectionValid(event)) {
  state = Authenticating
  logger.info(state.toString)
  push(sslOut, authMessage)
} else {
  failStage(new BetfairConnectionFailedException(s"Connection to 
Betfair failed"))
}
  case Authenticating =>
val event = grab(sslIn)
logger.info(event.toString)
if (authenticationValid(event)) {
  state = Subscribing
  logger.info(state.toString)
  push(sslOut, marketSubscription)
} else {
  failStage(new 
BetfairAuthenticationFailedException(s"Authentication with Betfair failed"))
}
  case Subscribing =>
val event = grab(sslIn)
logger.info(event.toString)
if (subscriptionValid(event)) {
  state = Subscribed
  logger.info(state.toString)
} else {
  failStage(new BetfairSubscriptionFailedException(s"Subscription 
with Betfair failed"))
}
  case Subscribed =>
val event = grab(sslIn)
logger.info(event.toString)
if (heatbeatValid(event)) {
  logger.info("Heartbeat")
  pull(sslIn)
} else {
  push(bytesOut, event)
  logger.info("MarketChange")
}
}
  }

  override def onUpstreamFinish(): Unit = complete(bytesOut)
})

// I2
setHandler(bytesIn, new InHandler {
  override def onPush() = {
  }

  override def onUpstreamFinish(): Unit = complete(sslOut)
})

// Called when transport pull for data
// O1
setHandler(bytesOut, new OutHandler {
  override def onPull() = state match {
case Connecting =>
case Authenticating =>
case Subscribing =>
case Subscribed => pull(sslIn)
  }

  override def onDownstreamFinish(): Unit = cancel(sslIn)
})

// O2
setHandler(sslOut, new OutHandler {
  override def onPull() = state match {
case Connecting => pull(sslIn)
case Authenticating => pull(sslIn)
case Subscribing => pull(sslIn)
case _ =>
  }

  override def onDownstreamFinish(): Unit = cancel(bytesIn)
})

  }
}


This code works as expected in the 

[akka-user] Architectural suggestion on stream mixed Actor/Stream application

[Henrik Larsson]

Im finding the idea of Akka Stream a very nice fit to the application im 
building. Im building a realtime decision system with the following 
functionality.


   1. Make HTTPS POST request to get a *sessionToken* string
   2. Start a keep-alive service to ping session server at 3h intervals
   3. Use *sessionToken* as payload when in connection message when 
   connecting to TCP socket to start receive *DataItems*
   4. Create a stream pipeline from the socket connection as a function 
   from *DataItem => Decision*
   5. Handle session management faults and any faults in the pipeline

What I have so far is an Actor hierarchy with an FSM SessionManagerActor 
which contains the *sessionToken *state, it gets it from a child actor wich 
does the actual getSessionRequest, when SessionManagerActor receives the 
*sessionToken* it spawns a keep-alive actor.
The problem starts when I want to use the *TCP* flow from Akka stream to 
setup the stream. All the ways im trying to connect the sessionToken as 
input the the TCP stream seems like an ugly hack and I cant figure out how 
to do error handling if i for example need to update the session token.

What would be the best solution for a problem like this and should i use 
Akka Stream for all parts, even keep-alive actor etc. Any input is highly 
appreciated since i feel if i only get and understanding of this part of 
the problem I will be able to solve the actual decision support part.

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Confusion in documentation regarding HTTPS support in Akka HTTP

[Henrik Larsson]

Nice thanks I got it working now! However is there a way to set HTTPS 
settings on the Http().singleRequest? The docs says it uses the default but 
i just want to set the HTTPS setting on this speficic request, all others 
should have another setting.

On Monday, May 8, 2017 at 8:19:19 AM UTC+2, Konrad Malawski wrote:
>
> It's plain Java, configure the SSLContext as usual and provide it to the 
> HttpsConnectionContext.
>
> -- 
> Konrad `ktoso` Malawski
> Akka <http://akka.io> @ Lightbend <http://lightbend.com>
>
> On 8 May 2017 at 15:18:21, Henrik Larsson (favetel...@gmail.com 
> ) wrote:
>
> Ok thanks, so far I have found this example wich i guess i can make work 
> on the client side also 
> https://github.com/rklaehn/akkahttpsserver/blob/master/src/main/scala/httpsserver/Server.scala
> However I would like to send the key and crt separate and in the exampel 
> they use p12. Is there any example on how to configure with a case like 
> this?
>
> On Monday, May 8, 2017 at 8:09:26 AM UTC+2, Konrad Malawski wrote: 
>>
>> My guess right now is that they do the same thing but the part under 
>> SSL-Config sets up the system default HttpsContext, is this correct?
>>
>> Yes, that's the case.
>>
>> SSLConfig is merely a helper utility you can use to easily set up things. 
>> You can not use it if you don't want to.
>>
>>
>> -- 
>> Konrad `ktoso` Malawski
>> Akka <http://akka.io> @ Lightbend <http://lightbend.com>
>>
>> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ: 
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups 
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an 
> email to akka-user+...@googlegroups.com .
> To post to this group, send email to akka...@googlegroups.com 
> .
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
>

-- 
>>>>>>>>>>  Read the docs: http://akka.io/docs/
>>>>>>>>>>  Check the FAQ: 
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Confusion in documentation regarding HTTPS support in Akka HTTP

[Henrik Larsson]

Ok thanks, so far I have found this example wich i guess i can make work on 
the client side also 
https://github.com/rklaehn/akkahttpsserver/blob/master/src/main/scala/httpsserver/Server.scala
However I would like to send the key and crt separate and in the exampel 
they use p12. Is there any example on how to configure with a case like 
this?

On Monday, May 8, 2017 at 8:09:26 AM UTC+2, Konrad Malawski wrote:
>
> My guess right now is that they do the same thing but the part under 
> SSL-Config sets up the system default HttpsContext, is this correct?
>
> Yes, that's the case.
>
> SSLConfig is merely a helper utility you can use to easily set up things. 
> You can not use it if you don't want to.
>
>
> -- 
> Konrad `ktoso` Malawski
> Akka  @ Lightbend 
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.