BitcoinJ

BitcoinJ.scala

package actors

import java.io.File
import java.util
import java.util.concurrent.TimeUnit

import akka.actor.Actor
import com.google.common.util.concurrent.{FutureCallback, Futures}
import org.bitcoinj.core._
import org.bitcoinj.net.discovery.DnsDiscovery
import org.bitcoinj.params.MainNetParams
import org.bitcoinj.store.{SPVBlockStore, H2FullPrunedBlockStore}
import org.bitcoinj.utils.BriefLogFormatter
import org.joda.time.DateTime

import scala.collection.concurrent.TrieMap
import scala.collection.JavaConversions._
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.Try


class BitcoinjActor extends Actor {
  import BitcoinjActor._

  override def receive = {
    case BitcoinJInit =>
      //BriefLogFormatter.init()

      peerGroup.addPeerDiscovery(new DnsDiscovery(params))
      peerGroup.addEventListener(peerEventListener)
      peerGroup.startAsync()
      peerGroup.addWallet(new Wallet(params))
      peerGroup.downloadBlockChain()


    case StartWatching(address) =>
      Future {
        println(s"Going to watch over $address")
        val addr = new Address(params, address)
        val wallet = new Wallet(params)
        wallet.addWatchedAddress(addr, DateTime.now().minusHours(1).getMillis)
        wallet.addEventListener(generateWalletEventListener(wallet))
        wallet.autosaveToFile(new File(s"/tmp/bitcoinj/wallets/w_13_$address"), 1, TimeUnit.SECONDS, null)
        wallets += address -> wallet

        println("Conneted peers: " + peerGroup.getConnectedPeers.size())
        peerGroup.addWallet(wallet)

        //workaround for BitcoinJ v 0.12.x, should be fixed with 0.13
        //https://github.com/bitcoinj/bitcoinj/pull/239
        peerGroup.getConnectedPeers.foreach(_.addWallet(wallet))
      }

    case GetTransactions(address) =>
      println(s"$address balance: " + wallets.get(address).get.getWatchedBalance.toFriendlyString)
      println(s"$address transactions: " + wallets.get(address).get.getTransactionsByTime)
      println(s"$address pending transactions: " + wallets.get(address).get.getPendingTransactions)
  }

  def generateWalletEventListener(trackedWallet:Wallet) = new AbstractWalletEventListener() {
    override def onCoinsReceived(wallet: Wallet, tx:Transaction, prevBalance:Coin, newBalance:Coin) {
      tx.getOutputs.map { to =>
        if (to.isMineOrWatched(trackedWallet)) {
          try {
            val toAddress = to.getScriptPubKey.getToAddress(params).toString
            println(tx.getUpdateTime + " onCoinsReceived for "
              + toAddress + " " + tx.getValueSentToMe(wallet).toFriendlyString)
            println("new balance: " + newBalance.toFriendlyString)
          } catch {
            case e: ScriptException =>
              println("can't fetch destination address from transactionOutput: " + e.getLocalizedMessage)
          }
        }
      }
    }

    override def onCoinsSent(wallet: Wallet, tx: Transaction, prevBalance: Coin, newBalance: Coin): Unit = {
      tx.getOutputs.map { to =>
        if (to.isMineOrWatched(trackedWallet)) {
          try {
            val toAddress = to.getScriptPubKey.getToAddress(params).toString
            println(tx.getUpdateTime + " onCoinsSent for "
              + toAddress + " " + tx.getValueSentToMe(wallet).toFriendlyString)
            println("new balance: " + newBalance.toFriendlyString)
          } catch {
            case e: ScriptException =>
              println("can't fetch destination address from transactionOutput: " + e.getLocalizedMessage)
          }
        }
      }
    }
  }

  object peerEventListener extends PeerEventListener {
    override def onBlocksDownloaded(peer: Peer, block: Block, blocksLeft: Int){}

    override def onChainDownloadStarted(peer:Peer, blocksLeft:Int){}

    override def onPeerConnected(peer:Peer, peerCount:Int) =
      println(s"peer connected ${peer.getAddress.getAddr.getHostAddress} peerCount $peerCount")

    override def onPeerDisconnected(peer:Peer, peerCount:Int) =
      println(s"peer disconnected ${peer.getAddress.getAddr.getHostAddress} peerCount $peerCount")

    override def onPreMessageReceived(peer:Peer, m:Message) = null

    override def onTransaction(peer:Peer, t:Transaction) { }

    override def getData(peer:Peer, m:GetDataMessage) = null

    override def onPeersDiscovered(set: util.Set[PeerAddress]): Unit = {}
  }
}


object BitcoinjActor {
  lazy val wallets = TrieMap[String, Wallet]()

  BriefLogFormatter.init()

  private lazy val params = MainNetParams.get()
  private lazy val blockStore = new H2FullPrunedBlockStore(params, "/tmp/bstore_13", 1000)
  private lazy val peerGroup = new PeerGroup(params, new BlockChain(params, blockStore))


  case object BitcoinJInit
  case class StartWatching(address:String)
  case class GetTransactions(address:String)
}



object BitcoinjTester {
  lazy val params = MainNetParams.get()
  lazy val wallet = new Wallet(params)
  
  def main(args:Array[String]): Unit ={

    val bitcoinAgent = actorSystem.actorOf(Props[BitcoinjActor]())
    bitcoinAgent ! BitcoinjActor.BitcoinJInit
    Thread.sleep(2000)
    bitcoinAgent ! BitcoinjActor.StartWatching("18KLcRV6imKWWGbT9riSquxcZjnpkuAVCX")
    bitcoinAgent ! BitcoinjActor.StartWatching("14hRZvN8Z3FeD97dw6jeXmd4FcZ6o3g4De")
    bitcoinAgent ! BitcoinjActor.StartWatching("1MCxXN7inBqHjgbNW3Kzq4GRc6h843TRFA")

    actorSystem.scheduler.schedule(1.minute, 10.seconds) {
      bitcoinAgent ! BitcoinjActor.GetTransactions("1MCxXN7inBqHjgbNW3Kzq4GRc6h843TRFA")
    }
  }
}