Removing entities from DB after X mins...

[Emil H]

Hi!

For fun I have started to develop a very basic bittorrent tracker
using Lift while I'm on parental leave. So far so good, it manages to
share peer information and I can upload/download and search for
torrent files.

However, if a peer stop their client ungracefully and a stop event is
not sent to my tracker the peer will continue to appear as connected.
I think I read somewhere that it is possible to add a delete method
that is run after X amount of time, but I don't remember where. Is
this possible using some built-in feature of vanilla Lift? If so,
anyone got a short how-to?

If anyone is interested I'll put the code on github, when I've cleaned
it up a little...

Cheers,
  Emil Hellman

--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups "Lift" group.
To post to this group, send email to liftweb@...
To unsubscribe from this group, send email to liftweb+unsubscribe@...
For more options, visit this group at http://groups.google.com/group/liftweb?hl=en
-~----------~----~----~----~------~----~------~--~---

[bearfeeder]

On Fri, Nov 6, 2009 at 1:15 AM, Emil <emil.hellman@...> wrote:

Hi!

For fun I have started to develop a very basic bittorrent tracker
using Lift while I'm on parental leave. So far so good, it manages to
share peer information and I can upload/download and search for
torrent files.

However, if a peer stop their client ungracefully and a stop event is
not sent to my tracker the peer will continue to appear as connected.
I think I read somewhere that it is possible to add a delete method
that is run after X amount of time, but I don't remember where. Is
this possible using some built-in feature of vanilla Lift? If so,
anyone got a short how-to?

This is probably more elaborate than you want, but it's how Lift manages sessions.  You can find the code in LiftSession.scala:

private[http] case class AddSession(session: LiftSession)
private[http] case class RemoveSession(sessionId: String)
case class SessionWatcherInfo(sessions: Map[String, LiftSession])

/**
 * Manages LiftSessions because the servlet container is less than optimal at
 * timing sessions out.
 */
object SessionMaster extends LiftActor {
  private var sessions: Map[String, LiftSession] = Map.empty
  private object CheckAndPurge

  def getSession(id: String, otherId: Box[String]): Box[LiftSession] = synchronized {
    otherId.flatMap(sessions.get) or Box(sessions.get(id))
  }

  /**
   * Put an Actor in this list and the Actor will receive a message
   * every 10 seconds with the current list of sessions:
   * SessionWatcherInfo
   */
  @volatile var sessionWatchers: List[LiftActor] = Nil

  /**
   * Returns a LiftSession or Empty if not found
   */
  def getSession(httpSession: => HTTPSession, otherId: Box[String]): Box[LiftSession] =
    synchronized {
      otherId.flatMap(sessions.get) or Box(sessions.get(httpSession.sessionId))
    }

  /**
   * Returns a LiftSession or Empty if not found
   */
  def getSession(req: HTTPRequest, otherId: Box[String]): Box[LiftSession] =
    synchronized {
      otherId.flatMap(sessions.get) or Box(sessions.get(req.session.sessionId))
    }

  /**
   * Adds a new session to SessionMaster
   */
  def addSession(liftSession: LiftSession) {
    synchronized {
      sessions = sessions + (liftSession.uniqueId -> liftSession)
    }
    liftSession.startSession()
    liftSession.httpSession.foreach(_.link(liftSession))
  }

  protected def messageHandler = reaction

  private val reaction: PartialFunction[Any, Unit] = {
    case RemoveSession(sessionId) =>
      val ses = synchronized(sessions)
      ses.get(sessionId).foreach {
        s =>
                try {
                  s.doShutDown
                  try {
                    s.httpSession.foreach(_.unlink(s))
                  } catch {
                    case e => // ignore... sometimes you can't do this and it's okay
                  }
                } catch {
                  case e => Log.error("Failure in remove session", e)

                } finally {
                  synchronized {sessions = sessions - sessionId}
                }
      }

    case CheckAndPurge =>
      val now = millis
      val ses = synchronized {sessions}
      for ((id, session) <- ses.elements) {
        session.doCometActorCleanup()
        if (now - session.lastServiceTime > session.inactivityLength || session.markedForTermination) {
          Log.info(" Session " + id + " expired")
          this.sendMsg(RemoveSession(id))
        } else {
          session.cleanupUnseenFuncs()
        }
      }
      if (!Props.inGAE) {
        sessionWatchers.foreach(_ ! SessionWatcherInfo(ses))
        doPing()
      }
  }


  private[http] def sendMsg(in: Any): Unit =
    if (!Props.inGAE) this ! in
    else {
      this.synchronized {
        tryo {
          if (reaction.isDefinedAt(in)) reaction.apply(in)
        }
      }
    }

  private def doPing() {
    if (!Props.inGAE) {
      try {
        ActorPing schedule (this, CheckAndPurge, 10 seconds)
      } catch {
        case e => Log.error("Couldn't start SessionMaster ping", e)
      }
    }
  }
}

 Hope this helps.

If anyone is interested I'll put the code on github, when I've cleaned
it up a little...

Cheers,
 Emil Hellman

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Surf the harmonics

--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups "Lift" group.
To post to this group, send email to liftweb@...
To unsubscribe from this group, send email to liftweb+unsubscribe@...
For more options, visit this group at http://groups.google.com/group/liftweb?hl=en
-~----------~----~----~----~------~----~------~--~---

[Emil H]

Thanks!

this pointed me in the right direction, I think. Adding a LiftActor
that removes my Peer entity to sessionWatchers would solve part of the
problem I suppose. However, a corner case is when a single peer
changes it's ip for some reason and connects from the new IP address.
This would create a new session. I suppose I could create an actor
that gets a message each time a peer connects and if it doesn't
connect in time deletes the peer...

Would using thin approach be a good idea? Should the PeerMonitorActor
hook up to some Watcher that can restart it (similar to the Supervisor
behaviour in Erlang/OTP)? Is there any documentation on how to use
actors in lift? Should I write something on it? I'm sure the issue of
removing database entries unless something happens within a timeperiod
is a pretty common use case (e.g. registering an account on a site).

Cheers,
  Emil

On Nov 6, 5:45 pm, David Pollak <feeder.of.the.be...@...> wrote:

> On Fri, Nov 6, 2009 at 1:15 AM, Emil <emil.hell...@...> wrote:
>
> > Hi!
>
> > For fun I have started to develop a very basic bittorrent tracker
> > using Lift while I'm on parental leave. So far so good, it manages to
> > share peer information and I can upload/download and search for
> > torrent files.
>
> > However, if a peer stop their client ungracefully and a stop event is
> > not sent to my tracker the peer will continue to appear as connected.
> > I think I read somewhere that it is possible to add a delete method
> > that is run after X amount of time, but I don't remember where. Is
> > this possible using some built-in feature of vanilla Lift? If so,
> > anyone got a short how-to?
>
> This is probably more elaborate than you want, but it's how Lift manages
> sessions.  You can find the code in LiftSession.scala:
>
> private[http] case class AddSession(session: LiftSession)
> private[http] case class RemoveSession(sessionId: String)
> case class SessionWatcherInfo(sessions: Map[String, LiftSession])
>
> /**
>  * Manages LiftSessions because the servlet container is less than optimal
> at
>  * timing sessions out.
>  */
> object SessionMaster extends LiftActor {
>   private var sessions: Map[String, LiftSession] = Map.empty
>   private object CheckAndPurge
>
>   def getSession(id: String, otherId: Box[String]): Box[LiftSession] =
> synchronized {
>     otherId.flatMap(sessions.get) or Box(sessions.get(id))
>   }
>
>   /**
>    * Put an Actor in this list and the Actor will receive a message
>    * every 10 seconds with the current list of sessions:
>    * SessionWatcherInfo
>    */
>   @volatile var sessionWatchers: List[LiftActor] = Nil
>
>   /**
>    * Returns a LiftSession or Empty if not found
>    */
>   def getSession(httpSession: => HTTPSession, otherId: Box[String]):
> Box[LiftSession] =
>     synchronized {
>       otherId.flatMap(sessions.get) or
> Box(sessions.get(httpSession.sessionId))
>     }
>
>   /**
>    * Returns a LiftSession or Empty if not found
>    */
>   def getSession(req: HTTPRequest, otherId: Box[String]): Box[LiftSession] =
>     synchronized {
>       otherId.flatMap(sessions.get) or
> Box(sessions.get(req.session.sessionId))
>     }
>
>   /**
>    * Adds a new session to SessionMaster
>    */
>   def addSession(liftSession: LiftSession) {
>     synchronized {
>       sessions = sessions + (liftSession.uniqueId -> liftSession)
>     }
>     liftSession.startSession()
>     liftSession.httpSession.foreach(_.link(liftSession))
>   }
>
>   protected def messageHandler = reaction
>
>   private val reaction: PartialFunction[Any, Unit] = {
>     case RemoveSession(sessionId) =>
>       val ses = synchronized(sessions)
>       ses.get(sessionId).foreach {
>         s =>
>                 try {
>                   s.doShutDown
>                   try {
>                     s.httpSession.foreach(_.unlink(s))
>                   } catch {
>                     case e => // ignore... sometimes you can't do this and
> it's okay
>                   }
>                 } catch {
>                   case e => Log.error("Failure in remove session", e)
>
>                 } finally {
>                   synchronized {sessions = sessions - sessionId}
>                 }
>       }
>
>     case CheckAndPurge =>
>       val now = millis
>       val ses = synchronized {sessions}
>       for ((id, session) <- ses.elements) {
>         session.doCometActorCleanup()
>         if (now - session.lastServiceTime > session.inactivityLength ||
> session.markedForTermination) {
>           Log.info(" Session " + id + " expired")
>           this.sendMsg(RemoveSession(id))
>         } else {
>           session.cleanupUnseenFuncs()
>         }
>       }
>       if (!Props.inGAE) {
>         sessionWatchers.foreach(_ ! SessionWatcherInfo(ses))
>         doPing()
>       }
>   }
>
>   private[http] def sendMsg(in: Any): Unit =
>     if (!Props.inGAE) this ! in
>     else {
>       this.synchronized {
>         tryo {
>           if (reaction.isDefinedAt(in)) reaction.apply(in)
>         }
>       }
>     }
>
>   private def doPing() {
>     if (!Props.inGAE) {
>       try {
>         ActorPing schedule (this, CheckAndPurge, 10 seconds)
>       } catch {
>         case e => Log.error("Couldn't start SessionMaster ping", e)
>       }
>     }
>   }
>
> }
>
>  Hope this helps.
>
>
>
> > If anyone is interested I'll put the code on github, when I've cleaned
> > it up a little...
>
> > Cheers,
> >  Emil Hellman
>
> --
> Lift, the simply functional web frameworkhttp://liftweb.net
> Beginning Scalahttp://www.apress.com/book/view/1430219890
> Follow me:http://twitter.com/dpp
> Surf the harmonics

--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups "Lift" group.
To post to this group, send email to liftweb@...
To unsubscribe from this group, send email to liftweb+unsubscribe@...
For more options, visit this group at http://groups.google.com/group/liftweb?hl=en
-~----------~----~----~----~------~----~------~--~---