client.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821
  1. package main
  2. /*
  3. #cgo CFLAGS: -O2
  4. #cgo LDFLAGS: -lcrypto -lm
  5. #include "../c/dpf.h"
  6. #include "../c/okvClient.h"
  7. #include "../c/dpf.c"
  8. #include "../c/okvClient.c"
  9. */
  10. import "C"
  11. //sssssssssssss
  12. import (
  13. "2PPS/lib"
  14. "bufio"
  15. "bytes"
  16. "crypto/rand"
  17. "crypto/sha256"
  18. "crypto/tls"
  19. "encoding/json"
  20. "fmt"
  21. "math/big"
  22. mr "math/rand"
  23. "net"
  24. "os"
  25. "sort"
  26. "strconv"
  27. "strings"
  28. "sync"
  29. "time"
  30. "unsafe"
  31. "golang.org/x/crypto/nacl/box"
  32. )
  33. type tweet struct {
  34. Topics []string
  35. Text string
  36. }
  37. const leader string = "127.0.0.1:4441"
  38. //needs to be changed at leader/follower/client at the same time
  39. const numClients = 2000
  40. //mylimit=8000
  41. //sudo prlimit --nofile=$mylimit --pid $$; ulimit -n $mylimit
  42. const dataLength int = 256
  43. const numThreads int = 12
  44. //Maximum Transport Unit
  45. const mtu int = 1100
  46. var dbWriteSize int
  47. var round int
  48. var topicList []string
  49. var archiveTopicList []string
  50. var neededSubscriptions int
  51. var maxTimePerRound time.Duration = 30 * time.Second
  52. var startTime int
  53. //todo! expand this for multiple clients
  54. var archiveInterests = make([]int, 1)
  55. var sharedSecret [numClients][2][32]byte = createSharedSecret()
  56. var wantsArchive = make([]byte, 1)
  57. var leaderPublicKey *[32]byte
  58. var followerPublicKey *[32]byte
  59. var clientPrivateKey [numClients]*[32]byte
  60. var clientPublicKey [numClients]*[32]byte
  61. func main() {
  62. wg := &sync.WaitGroup{}
  63. for i := 0; i < numClients; i++ {
  64. wg.Add(1)
  65. go client(i)
  66. time.Sleep(1 * time.Millisecond)
  67. }
  68. wg.Wait()
  69. }
  70. func client(clientNumber int) {
  71. generatedPublicKey, generatedPrivateKey, err := box.GenerateKey(rand.Reader)
  72. if err != nil {
  73. panic(err)
  74. }
  75. clientPrivateKey[clientNumber] = generatedPrivateKey
  76. clientPublicKey[clientNumber] = generatedPublicKey
  77. C.initializeCipher()
  78. //initializes the connection to the leader
  79. conf := &tls.Config{
  80. InsecureSkipVerify: true,
  81. }
  82. leaderConn, err := tls.Dial("tcp", leader, conf)
  83. if err != nil {
  84. fmt.Println("clientNumber", clientNumber)
  85. panic(err)
  86. }
  87. leaderConn.SetDeadline(time.Time{})
  88. //receives topics first so client can participate asap
  89. receiveTopicLists(leaderConn)
  90. //gets the public keys of both servers
  91. var tmpLeaderPubKey [32]byte
  92. _, err = leaderConn.Read(tmpLeaderPubKey[:])
  93. if err != nil {
  94. panic(err)
  95. }
  96. leaderPublicKey = &tmpLeaderPubKey
  97. var tmpFollowerPubKey [32]byte
  98. _, err = leaderConn.Read(tmpFollowerPubKey[:])
  99. if err != nil {
  100. panic(err)
  101. }
  102. followerPublicKey = &tmpFollowerPubKey
  103. //sends own public key
  104. writeTo(leaderConn, clientPublicKey[clientNumber][:])
  105. neededSubscriptionsBytes := readFrom(leaderConn, 4)
  106. neededSubscriptions = byteToInt(neededSubscriptionsBytes)
  107. startTimeBytes := readFrom(leaderConn, 4)
  108. startTime = byteToInt(startTimeBytes)
  109. //setup ends above
  110. //while client is active he is always connected and has to participate
  111. for {
  112. //gets current phase
  113. phase := readFrom(leaderConn, 1)
  114. if phase[0] == 1 {
  115. //gets current dbWriteSize from leader
  116. dbWriteSizeBytes := readFrom(leaderConn, 4)
  117. dbWriteSize = byteToInt(dbWriteSizeBytes)
  118. //todo! put into tweet creation
  119. //roundAsBytes := readFrom(leaderConn, 4)
  120. roundAsBytes := make([]byte, 4)
  121. _, err = leaderConn.Read(roundAsBytes)
  122. if err != nil {
  123. panic(err)
  124. }
  125. round = byteToInt(roundAsBytes)
  126. if clientNumber == 0 {
  127. fmt.Println("Round ", round)
  128. }
  129. //request virtualAddress from leader via pirQuery
  130. encryptedQueryLeader, encryptedQueryFollower := createAuditPIRQuery(clientNumber)
  131. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, false)
  132. pos := receiveVirtualAddress(sharedSecret[clientNumber], leaderConn)
  133. tweet := getRealTweet(clientNumber)
  134. //prep the query
  135. dataSize := len(tweet)
  136. querySize := make([]byte, 4)
  137. cQuerySize := C.int(byteToInt(querySize))
  138. var dpfQueryA *C.uchar
  139. var dpfQueryB *C.uchar
  140. C.prepQuery(C.int(pos), C.int(dbWriteSize), (*C.uchar)(&tweet[0]), C.int(dataSize), &cQuerySize, &dpfQueryA, &dpfQueryB)
  141. intQuerySize := int(cQuerySize) //byteToInt(querySize)
  142. //write the query
  143. queryAPlaintext := C.GoBytes(unsafe.Pointer(dpfQueryA), C.int(intQuerySize))
  144. //encrypts queryA and appends it to message
  145. var nonce [24]byte
  146. //fill nonce with randomness
  147. _, err = rand.Read(nonce[:])
  148. if err != nil {
  149. panic("couldn't get randomness for nonce!")
  150. }
  151. dpfQueryAEncrypted := box.Seal(nonce[:], queryAPlaintext, &nonce, leaderPublicKey, clientPrivateKey[clientNumber])
  152. //encrypts queryB and appends it to message
  153. queryBPlaintext := C.GoBytes(unsafe.Pointer(dpfQueryB), C.int(intQuerySize))
  154. //fill nonce with randomness
  155. _, err = rand.Read(nonce[:])
  156. if err != nil {
  157. panic("couldn't get randomness for nonce!")
  158. }
  159. dpfQueryBEncrypted := box.Seal(nonce[:], queryBPlaintext, &nonce, followerPublicKey, clientPrivateKey[clientNumber])
  160. //writes the dpfQuery to the leader
  161. dpfLengthBytes := intToByte(len(dpfQueryAEncrypted))
  162. writeTo(leaderConn, dpfLengthBytes)
  163. writeTo(leaderConn, dpfQueryAEncrypted)
  164. writeTo(leaderConn, dpfQueryBEncrypted)
  165. C.free(unsafe.Pointer(dpfQueryA))
  166. C.free(unsafe.Pointer(dpfQueryB))
  167. } else if phase[0] == 3 {
  168. /*
  169. possible Values
  170. 0 : new client
  171. leader expects sharedSecrets, expects pirQuery
  172. 1 : update needed
  173. leader sends topicList, performs local update of sharedSecret, expects pirQuery
  174. 2 : no update needed
  175. nothing
  176. */
  177. subPhase := readFrom(leaderConn, 1)
  178. var encryptedQueryLeader, encryptedQueryFollower []byte
  179. //first time participating
  180. if subPhase[0] == 0 {
  181. receiveTopicLists(leaderConn)
  182. encryptedQueryLeader, encryptedQueryFollower = createPIRQuery(int(subPhase[0]), clientNumber)
  183. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, false)
  184. }
  185. //updates the topic list and what client is interested in
  186. if subPhase[0] == 1 {
  187. receiveTopicLists(leaderConn)
  188. //updates local secret
  189. for index := 0; index < 2; index++ {
  190. sharedSecret[clientNumber][index] = sha256.Sum256(sharedSecret[clientNumber][index][:])
  191. }
  192. encryptedQueryLeader, encryptedQueryFollower = createPIRQuery(int(subPhase[0]), clientNumber)
  193. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, false)
  194. }
  195. receiveTweets(sharedSecret[clientNumber], leaderConn, false, clientNumber)
  196. if len(archiveTopicList) > 0 {
  197. wantsArchive[0] = 0 //archive test
  198. } else {
  199. wantsArchive[0] = 0
  200. }
  201. writeTo(leaderConn, wantsArchive)
  202. if wantsArchive[0] == 1 && len(archiveTopicList) > 0 {
  203. encryptedQueryLeader, encryptedQueryFollower = createPIRQuery(-1, clientNumber)
  204. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, true)
  205. receiveTweets(sharedSecret[clientNumber], leaderConn, true, clientNumber)
  206. }
  207. } else {
  208. fmt.Println("Phase", phase)
  209. panic("somethin went wrong")
  210. }
  211. }
  212. }
  213. //creates and sends the pirQuerys for each server
  214. func createPIRQuery(subPhase int, clientNumber int) ([]byte, []byte) {
  215. //later this will be taken from gui, this is only for testing
  216. topicsOFInterest := make([]int, 10)
  217. topicsOFInterest[0] = 0
  218. topicsOFInterest[1] = 1
  219. topicsOFInterest[9] = 1
  220. archiveInterests[0] = 1
  221. //todo! repeat for archive
  222. tmpNeededSubscriptions := neededSubscriptions
  223. if tmpNeededSubscriptions > len(topicList) {
  224. tmpNeededSubscriptions = len(topicList)
  225. }
  226. tmptopicsOfInterest := make([]int, len(topicList))
  227. copy(tmptopicsOfInterest, topicsOFInterest)
  228. tmpTopicList := make([]string, len(topicList))
  229. copy(tmpTopicList, topicList)
  230. if wantsArchive[0] == 1 && subPhase == -1 {
  231. tmpNeededSubscriptions = len(archiveInterests)
  232. if tmpNeededSubscriptions > len(archiveTopicList) {
  233. tmpNeededSubscriptions = len(archiveTopicList)
  234. }
  235. copy(tmptopicsOfInterest, archiveInterests) //archiveInterests from gui
  236. copy(tmpTopicList, archiveTopicList)
  237. }
  238. //creates fake topicsOfInterest if client is boooring
  239. if len(tmptopicsOfInterest) < tmpNeededSubscriptions && subPhase != -1 {
  240. tmptopicsOfInterest = addFakeInterests(len(tmpTopicList), tmptopicsOfInterest, false)
  241. }
  242. //pirQuery [topicsofinterest][serverAmount][topicAmount]byte
  243. pirQuerys := make([][][]byte, len(tmptopicsOfInterest))
  244. for i := range pirQuerys {
  245. pirQuerys[i] = make([][]byte, 2)
  246. for j := range pirQuerys[i] {
  247. pirQuerys[i][j] = make([]byte, len(tmpTopicList))
  248. }
  249. }
  250. //for leader
  251. //pirQuery will be filled with random bits
  252. for topic := range tmptopicsOfInterest {
  253. for index := range tmpTopicList {
  254. bit, err := rand.Int(rand.Reader, big.NewInt(2))
  255. if err != nil {
  256. panic(err)
  257. }
  258. pirQuerys[topic][0][index] = byte(bit.Int64())
  259. }
  260. }
  261. tmptopicsOfInterestBytes := make([]byte, tmpNeededSubscriptions)
  262. for index := range tmptopicsOfInterestBytes {
  263. if tmptopicsOfInterest[index] == 1 {
  264. tmptopicsOfInterestBytes[index] = 1
  265. }
  266. }
  267. for topic := range tmptopicsOfInterest {
  268. for index := range tmpTopicList {
  269. if topic == index {
  270. if pirQuerys[topic][0][index] == 1 {
  271. pirQuerys[topic][1][index] = 0
  272. } else {
  273. pirQuerys[topic][1][index] = 1
  274. }
  275. } else {
  276. if pirQuerys[topic][0][index] == 0 {
  277. pirQuerys[topic][1][index] = 0
  278. } else {
  279. pirQuerys[topic][1][index] = 1
  280. }
  281. }
  282. }
  283. }
  284. //flattens the querys to be able to send them more efficently
  285. messagesFlattened := make([][]byte, 2)
  286. //adds the sharedSecret to the first pirQuery when first time participating
  287. if subPhase == 0 {
  288. for server := 0; server < 2; server++ {
  289. messagesFlattened[server] = append(messagesFlattened[server], sharedSecret[clientNumber][server][:]...)
  290. }
  291. }
  292. for server := range messagesFlattened {
  293. for topic := range pirQuerys {
  294. messagesFlattened[server] = append(messagesFlattened[server], pirQuerys[topic][server]...)
  295. }
  296. }
  297. var nonce [24]byte
  298. _, err := rand.Read(nonce[:])
  299. if err != nil {
  300. panic("couldn't get randomness for nonce!")
  301. }
  302. encryptedQueryLeader := box.Seal(nonce[:], messagesFlattened[0], &nonce, leaderPublicKey, clientPrivateKey[clientNumber])
  303. _, err = rand.Read(nonce[:])
  304. if err != nil {
  305. panic("couldn't get randomness for nonce!")
  306. }
  307. encryptedQueryFollower := box.Seal(nonce[:], messagesFlattened[1], &nonce, followerPublicKey, clientPrivateKey[clientNumber])
  308. return encryptedQueryLeader, encryptedQueryFollower
  309. }
  310. func sendQuerys(encryptedQueryLeader, encryptedQueryFollower []byte, leaderConn net.Conn, getArchive bool) {
  311. encryptedLength := len(encryptedQueryLeader)
  312. //sends the pirQuerysLength to the leader
  313. writeTo(leaderConn, intToByte(encryptedLength))
  314. //sends the pirQuerys to the leader
  315. writeTo(leaderConn, encryptedQueryLeader)
  316. writeTo(leaderConn, encryptedQueryFollower)
  317. if getArchive {
  318. writeTo(leaderConn, intToByte(len(archiveInterests)))
  319. }
  320. }
  321. func receiveVirtualAddress(sharedSecret [2][32]byte, leaderConn net.Conn) int {
  322. virtualAddressByte := readFrom(leaderConn, 4)
  323. //xores the sharedSecret
  324. for h := 0; h < 2; h++ {
  325. for i := 0; i < 4; i++ {
  326. virtualAddressByte[i] = virtualAddressByte[i] ^ sharedSecret[h][i]
  327. }
  328. }
  329. return byteToInt(virtualAddressByte)
  330. }
  331. func receiveTweets(sharedSecret [2][32]byte, leaderConn net.Conn, getArchive bool, clientNumber int) {
  332. tmpNeededSubscriptions := neededSubscriptions
  333. if tmpNeededSubscriptions > len(topicList) {
  334. tmpNeededSubscriptions = len(topicList)
  335. }
  336. if getArchive {
  337. tmpNeededSubscriptions = len(archiveInterests)
  338. if tmpNeededSubscriptions > len(archiveTopicList) {
  339. tmpNeededSubscriptions = len(archiveTopicList)
  340. }
  341. }
  342. for i := 0; i < tmpNeededSubscriptions; i++ {
  343. //client receives tweets
  344. tweetsLengthBytes := readFrom(leaderConn, 4)
  345. tweetsLength := byteToInt(tweetsLengthBytes)
  346. tweets := readFrom(leaderConn, tweetsLength)
  347. //fmt.Println(tweets[:10])
  348. //expand sharedSecret so it is of right length
  349. expandBy := len(tweets) / 32
  350. expandedSharedSecrets := make([][]byte, 2)
  351. for i := 0; i < 2; i++ {
  352. for j := 0; j < expandBy; j++ {
  353. expandedSharedSecrets[i] = append(expandedSharedSecrets[i], sharedSecret[i][:]...)
  354. }
  355. }
  356. //xors the received messge into the message to display
  357. for i := 0; i < 2; i++ {
  358. lib.Xor(expandedSharedSecrets[i][:], tweets)
  359. }
  360. //fmt.Println("PubKey", clientPublicKey[clientNumber], "Bytes", tweets)
  361. index := strings.Index(string(tweets), ";;")
  362. if index != -1 && clientNumber == 0 {
  363. //fmt.Println("Round", round, "Tweets length", len(tweets))
  364. fmt.Println("Correct")
  365. textArr := strings.Split(string(tweets), ";;;;;;;;")
  366. text := string(tweets)[:len(textArr)-1]
  367. fmt.Println("Round", round, "Text", text[:5], "Length", len(tweets))
  368. } else if index == -1 {
  369. fmt.Println("error")
  370. fmt.Println("round", round, "text:", string(tweets), "length", len(tweets))
  371. return
  372. //panic("received text not of correct format")
  373. }
  374. }
  375. }
  376. //creates a shared secret for each server
  377. func createSharedSecret() [numClients][2][32]byte {
  378. var tmpSharedSecret [numClients][2][32]byte
  379. for i := 0; i < numClients; i++ {
  380. for j := 0; j < 2; j++ {
  381. _, err := rand.Read(tmpSharedSecret[i][j][:])
  382. if err != nil {
  383. panic("couldn't get randomness for sharedSecret!")
  384. }
  385. }
  386. }
  387. return tmpSharedSecret
  388. }
  389. func createAuditPIRQuery(clientNumber int) ([]byte, []byte) {
  390. //pirQuery [serverAmount][dbWriteSize]byte
  391. pirQuerys := make([][]byte, 2)
  392. for i := range pirQuerys {
  393. pirQuerys[i] = make([]byte, dbWriteSize)
  394. }
  395. //for leader
  396. //pirQuery will be filled with random bits
  397. for index := range pirQuerys[0] {
  398. bit := mr.Intn(2)
  399. pirQuerys[0][index] = byte(bit)
  400. }
  401. copy(pirQuerys[1], pirQuerys[0])
  402. //the positon the virtual address will be taken from
  403. pos := mr.Intn(dbWriteSize)
  404. pirQuerys[0][pos] = 1
  405. pirQuerys[1][pos] = 0
  406. //flattens the querys to be able to send them more efficently
  407. messagesFlattened := make([][]byte, 2)
  408. //adds the sharedSecret to the pirQuery
  409. for server := 0; server < 2; server++ {
  410. messagesFlattened[server] = append(messagesFlattened[server], sharedSecret[clientNumber][server][:]...)
  411. }
  412. for server := 0; server < 2; server++ {
  413. messagesFlattened[server] = append(messagesFlattened[server], pirQuerys[server][:]...)
  414. }
  415. var nonce [24]byte
  416. _, err := rand.Read(nonce[:])
  417. if err != nil {
  418. panic("couldn't get randomness for nonce!")
  419. }
  420. encryptedQueryLeader := box.Seal(nonce[:], messagesFlattened[0], &nonce, leaderPublicKey, clientPrivateKey[clientNumber])
  421. _, err = rand.Read(nonce[:])
  422. if err != nil {
  423. panic("couldn't get randomness for nonce!")
  424. }
  425. encryptedQueryFollower := box.Seal(nonce[:], messagesFlattened[1], &nonce, followerPublicKey, clientPrivateKey[clientNumber])
  426. return encryptedQueryLeader, encryptedQueryFollower
  427. }
  428. //generates a topicOfInterest array with random values
  429. func addFakeInterests(max int, topicsOfInterest []int, doAuditing bool) []int {
  430. tmpNeededSubscriptions := neededSubscriptions
  431. if tmpNeededSubscriptions > len(topicList) {
  432. tmpNeededSubscriptions = len(topicList)
  433. }
  434. fakeTopicsOfInterest := make([]int, tmpNeededSubscriptions)
  435. maxInt := max
  436. //fills the array with unique random ascending values ranging from 0 to max
  437. for i := 0; i < tmpNeededSubscriptions; i++ {
  438. fakeTopicsOfInterest[i] = mr.Intn(maxInt)
  439. for j := 0; j < i; j++ {
  440. if fakeTopicsOfInterest[i] == fakeTopicsOfInterest[j] {
  441. i--
  442. break
  443. }
  444. }
  445. }
  446. if doAuditing {
  447. sort.Ints(fakeTopicsOfInterest)
  448. return fakeTopicsOfInterest
  449. }
  450. //adds unique and new random numbers to topicOfInterests until length is satisfied
  451. for _, number := range fakeTopicsOfInterest {
  452. if !inList(number, topicsOfInterest) {
  453. topicsOfInterest = append(topicsOfInterest, number)
  454. }
  455. if len(topicsOfInterest) == tmpNeededSubscriptions {
  456. break
  457. }
  458. }
  459. sort.Ints(topicsOfInterest)
  460. return topicsOfInterest
  461. }
  462. func inList(number int, list []int) bool {
  463. for _, element := range list {
  464. if element == number {
  465. return true
  466. }
  467. }
  468. return false
  469. }
  470. func receiveTopicLists(leaderConn net.Conn) {
  471. for i := 0; i < 2; i++ {
  472. topicListLength := readFrom(leaderConn, 4)
  473. recTopicList := readFrom(leaderConn, byteToInt(topicListLength))
  474. var tmpTopicList []string
  475. arrayReader := bytes.NewReader(recTopicList[:])
  476. json.NewDecoder(arrayReader).Decode(&tmpTopicList)
  477. if i == 0 {
  478. topicList = tmpTopicList
  479. } else {
  480. archiveTopicList = tmpTopicList
  481. }
  482. }
  483. }
  484. func getRealTweet(clientNumber int) []byte {
  485. fUserList, err := os.Open("/home/simon/goCode/tweets/userList")
  486. if err != nil {
  487. panic(err)
  488. }
  489. defer fUserList.Close()
  490. currentLine := 0
  491. scanner := bufio.NewScanner(fUserList)
  492. userID := ""
  493. for scanner.Scan() {
  494. if currentLine == clientNumber {
  495. userID = scanner.Text()
  496. break
  497. }
  498. currentLine++
  499. }
  500. if userID == "" {
  501. panic("no userID picked")
  502. }
  503. fTweets, err := os.Open("/home/simon/goCode/tweets/userTweets/" + userID)
  504. if err != nil {
  505. panic(err)
  506. }
  507. defer fTweets.Close()
  508. scanner = bufio.NewScanner(fTweets)
  509. lowerBound := int(time.Now().Unix())
  510. upperBound := lowerBound + int(3*maxTimePerRound.Seconds()) + 10
  511. var tweet []byte
  512. for scanner.Scan() {
  513. lineArr := strings.Split(scanner.Text(), ", \"hashtags\"")
  514. lineArr = strings.Split(lineArr[0], ": ")
  515. lineArr = strings.Split(lineArr[1], " \"")
  516. timestamp, _ := strconv.Atoi(lineArr[0])
  517. //transforms timestamp to current time
  518. timestamp -= 1351742400
  519. timestamp += startTime
  520. if timestamp > lowerBound && timestamp < upperBound {
  521. lineArr = strings.Split(scanner.Text(), "[\"")
  522. line := lineArr[1]
  523. lineArr = strings.Split(line, "\"]")
  524. line = lineArr[0]
  525. lineArr = strings.Split(line, ",")
  526. line = strings.Join(lineArr, "")
  527. topicLine := strings.Split(line, "\"")
  528. var topics []byte
  529. for index, topic := range topicLine {
  530. if index%2 == 1 {
  531. continue
  532. }
  533. if len(topics)+len(topic) > dataLength-10 {
  534. break
  535. }
  536. topics = append(topics, []byte(topic)[:]...)
  537. topics = append(topics, []byte(",")[0])
  538. }
  539. topics = topics[:len(topics)-1]
  540. tweet = append(tweet, topics...)
  541. tweet = append(tweet, []byte(";")[0])
  542. r := mr.New(mr.NewSource(time.Now().UnixNano()))
  543. num := r.Intn(10000)
  544. if num == 0 {
  545. num = 1
  546. }
  547. tweet = append(tweet, []byte(strconv.Itoa(num) + ";;")[:]...)
  548. //fmt.Println("tweet", string(tweet))
  549. //adds padding
  550. length := dataLength - len(tweet)
  551. padding := make([]byte, length)
  552. rand.Read(padding)
  553. tweet = append(tweet, padding...)
  554. return tweet
  555. }
  556. }
  557. tweet = make([]byte, dataLength)
  558. return tweet
  559. }
  560. func getRandomTweet(clientNumber int) []byte {
  561. var tweet []byte
  562. r := mr.New(mr.NewSource(time.Now().UnixNano()))
  563. maxTopics := r.Intn(6)
  564. if maxTopics == 0 {
  565. maxTopics = 1
  566. }
  567. maxInt := 10000
  568. topicNumbers := make([]int, maxTopics)
  569. //fills the array with unique random ascending values ranging from 0 to maxInt
  570. for i := 0; i < maxTopics; i++ {
  571. topicNumbers[i] = mr.Intn(maxInt)
  572. for j := 0; j < i; j++ {
  573. if topicNumbers[i] == topicNumbers[j] {
  574. i--
  575. break
  576. }
  577. }
  578. }
  579. sort.Ints(topicNumbers)
  580. //fmt.Println("topicNumbers", topicNumbers)
  581. var topics []byte
  582. topicIndex := 0
  583. for i := 0; i < len(topicNumbers)*2; i++ {
  584. if i%2 == 0 {
  585. topics = append(topics, byte(topicNumbers[topicIndex]))
  586. topicIndex++
  587. } else if i != (len(topicNumbers)*2)-1 {
  588. topics = append(topics, []byte(",")[0])
  589. }
  590. }
  591. topics = append(topics, []byte(";")[0])
  592. num := r.Intn(100)
  593. if num == 0 {
  594. num = 1
  595. }
  596. text := []byte(strconv.Itoa(num) + ";")
  597. tweet = append(tweet, topics...)
  598. tweet = append(tweet, text...)
  599. tweet = append(tweet, []byte(";")[0])
  600. //fmt.Println("writing", string(text))
  601. //fmt.Println(topicNumbers)
  602. if len(tweet) > 32 {
  603. fmt.Println("lenlen", len(tweet))
  604. }
  605. //adds padding
  606. length := dataLength - len(tweet)
  607. padding := make([]byte, length)
  608. rand.Read(padding)
  609. tweet = append(tweet, padding...)
  610. return tweet
  611. }
  612. //sends the array to the connection
  613. func writeTo(connection net.Conn, array []byte) {
  614. remainingLength := len(array)
  615. for remainingLength > 0 {
  616. if remainingLength >= mtu {
  617. _, err := connection.Write(array[:mtu])
  618. if err != nil {
  619. panic(err)
  620. }
  621. array = array[mtu:]
  622. remainingLength -= mtu
  623. } else {
  624. _, err := connection.Write(array)
  625. if err != nil {
  626. panic(err)
  627. }
  628. remainingLength = 0
  629. }
  630. }
  631. }
  632. //reads an array which is returned and of size "size" from the connection
  633. func readFrom(connection net.Conn, size int) []byte {
  634. var array []byte
  635. remainingSize := size
  636. for remainingSize > 0 {
  637. var err error
  638. toAppend := make([]byte, mtu)
  639. if remainingSize > mtu {
  640. _, err = connection.Read(toAppend)
  641. array = append(array, toAppend...)
  642. remainingSize -= mtu
  643. } else {
  644. _, err = connection.Read(toAppend[:remainingSize])
  645. array = append(array, toAppend[:remainingSize]...)
  646. remainingSize = 0
  647. }
  648. if err != nil {
  649. panic(err)
  650. }
  651. }
  652. return array
  653. }
  654. func intToByte(myInt int) (retBytes []byte) {
  655. retBytes = make([]byte, 4)
  656. retBytes[3] = byte((myInt >> 24) & 0xff)
  657. retBytes[2] = byte((myInt >> 16) & 0xff)
  658. retBytes[1] = byte((myInt >> 8) & 0xff)
  659. retBytes[0] = byte(myInt & 0xff)
  660. return
  661. }
  662. func byteToInt(myBytes []byte) (x int) {
  663. x = int(myBytes[3])<<24 + int(myBytes[2])<<16 + int(myBytes[1])<<8 + int(myBytes[0])
  664. return
  665. }