Scala Cancel类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Scala›Scala教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Scala中akka.stream.actor.ActorPublisherMessage.Cancel类的典型用法代码示例。如果您正苦于以下问题：Scala Cancel类的具体用法？Scala Cancel怎么用？Scala Cancel使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了Cancel类的4个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Scala代码示例。

示例1: TweetPublisher

//设置package包名称以及导入依赖的类
package de.codecentric.dcos_intro

import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.{Cancel, Request}

class TweetPublisher extends ActorPublisher[Tweet] {
  override def receive: Receive = {
    case t: Tweet => {
      if (isActive && totalDemand > 0) {
        onNext(t)
      }
    }
    case Cancel => context.stop(self)
    case Request(_) => {}

  }
}

开发者ID:ftrossbach，项目名称:intro-to-dcos，代码行数:18，代码来源:TweetPublisher.scala

示例2: StreamingActor

//设置package包名称以及导入依赖的类
package wiii

import java.io.FileNotFoundException

import akka.actor.Props
import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.{Cancel, Request}
import akka.util.ByteString
import com.typesafe.scalalogging.LazyLogging
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{Path, FileSystem}


object StreamingActor {
    def props(host: String, port: Int, path: String) = Props(new StreamingActor(host, port, path))
}
class StreamingActor(host: String, port: Int, path: String) extends ActorPublisher[ByteString]  with LazyLogging {
    val filesys = {
        val conf = new Configuration()
        conf.set("fs.default.name", s"hdfs://$host:$port")
        FileSystem.get(conf)
    }

    val chunkSize = 1024
    val arr = Array.ofDim[Byte](chunkSize)

    def receive: Receive = {
        case Request(cnt) =>
            val uri = new Path(path)
            uri match {
                case p if !filesys.exists(p) => throw new FileNotFoundException(s"$p does not exist")
                case p if !filesys.getFileStatus(p).isFile => throw new FileNotFoundException(s"$p is not a file")
                case p =>
                    val is = filesys.open(p)
                    val readBytes = is.read(arr)
                    onNext(ByteString.fromArray(arr, 0, readBytes))
            }
        case Cancel => context.stop(self)
        case _ =>
    }
}

开发者ID:jw3，项目名称:example-hdfs-docker，代码行数:42，代码来源:StreamingActor.scala

示例3: ConnectionAgent

//设置package包名称以及导入依赖的类
package com.bisphone.sarf.implv1.tcpservice

import akka.actor.Props
import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.{Cancel, Request}
import com.bisphone.sarf.IOCommand
import org.slf4j.Logger

import scala.collection.mutable


private[implv1] class ConnectionAgent (logger: Logger) extends ActorPublisher[IOCommand] {

   val queue = mutable.Queue.empty[IOCommand]

   def tryDeliver (): Unit = {
      if (totalDemand > 0 && queue.nonEmpty) onNext(queue dequeue)
   }

   private def tryPush (cmd: IOCommand): Unit = {
      queue enqueue cmd
      tryDeliver()
   }

   def receive: Receive = {
      case cmd: IOCommand => tryPush(cmd)
      case Request(_) => tryDeliver()
      case Cancel =>
         if (logger.isDebugEnabled()) logger.debug(s"Actor($self) received 'Cancel' signal!")
         context stop self
   }

   override def preStart (): Unit = {
      if (logger.isDebugEnabled()) logger.debug(s"Actor($self) preStart")
   }

   override def postStop (): Unit = {
      if (logger.isDebugEnabled()) logger.debug(s"Actor($self) postStop")
   }

}

private[implv1] object ConnectionAgent {
   def props (logger: Logger) = Props {
      new ConnectionAgent(logger)
   }
}

开发者ID:bisphone，项目名称:SARF，代码行数:48，代码来源:ConnectionAgent.scala

示例4: NanoboardMessagePublisher

//设置package包名称以及导入依赖的类
package com.karasiq.nanoboard.server.streaming

import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.{Cancel, Request}
import com.karasiq.nanoboard.streaming.NanoboardEvent

import scala.annotation.tailrec

private[server] class NanoboardMessagePublisher extends ActorPublisher[NanoboardEvent] {
  override def preStart(): Unit = {
    super.preStart()
    context.system.eventStream.subscribe(self, classOf[NanoboardEvent])
  }

  override def postStop(): Unit = {
    context.system.eventStream.unsubscribe(self)
    super.postStop()
  }

  val maxBufferSize = 20
  var messageBuffer = Vector.empty[NanoboardEvent]

  override def receive: Receive = {
    case m: NanoboardEvent ?
      if (messageBuffer.isEmpty && totalDemand > 0) {
        onNext(m)
      } else {
        if (messageBuffer.length >= maxBufferSize) {
          messageBuffer = messageBuffer.tail :+ m
        } else {
          messageBuffer :+= m
        }
        deliverBuffer()
      }

    case Request(_) ?
      deliverBuffer()

    case Cancel ?
      context.stop(self)
  }

  @tailrec final def deliverBuffer(): Unit = {
    if (totalDemand > 0) {
      if (totalDemand <= Int.MaxValue) {
        val (use, keep) = messageBuffer.splitAt(totalDemand.toInt)
        messageBuffer = keep
        use foreach onNext
      } else {
        val (use, keep) = messageBuffer.splitAt(Int.MaxValue)
        messageBuffer = keep
        use foreach onNext
        deliverBuffer()
      }
    }
  }
}

开发者ID:Karasiq，项目名称:nanoboard，代码行数:58，代码来源:NanoboardMessagePublisher.scala

注：本文中的akka.stream.actor.ActorPublisherMessage.Cancel类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。