(data)
| 531 | |
| 532 | |
| 533 | def _checkAndShareBroadcastWithPeers(data): |
| 534 | if len(data) < 180: |
| 535 | logger.debug('The payload length of this broadcast packet is unreasonably low. Someone is probably trying funny business. Ignoring message.') |
| 536 | return |
| 537 | embeddedTime, = unpack('>Q', data[8:16]) |
| 538 | readPosition = 20 # bypass the nonce, time, and object type |
| 539 | broadcastVersion, broadcastVersionLength = decodeVarint( |
| 540 | data[readPosition:readPosition + 10]) |
| 541 | readPosition += broadcastVersionLength |
| 542 | if broadcastVersion >= 2: |
| 543 | streamNumber, streamNumberLength = decodeVarint(data[readPosition:readPosition + 10]) |
| 544 | readPosition += streamNumberLength |
| 545 | if not streamNumber in state.streamsInWhichIAmParticipating: |
| 546 | logger.debug('The streamNumber %s isn\'t one we are interested in.', streamNumber) |
| 547 | return |
| 548 | if broadcastVersion >= 3: |
| 549 | tag = data[readPosition:readPosition+32] |
| 550 | else: |
| 551 | tag = '' |
| 552 | inventoryHash = calculateInventoryHash(data) |
| 553 | if inventoryHash in Inventory(): |
| 554 | logger.debug('We have already received this broadcast object. Ignoring.') |
| 555 | return |
| 556 | # It is valid. Let's let our peers know about it. |
| 557 | objectType = 3 |
| 558 | Inventory()[inventoryHash] = ( |
| 559 | objectType, streamNumber, data, embeddedTime, tag) |
| 560 | # This object is valid. Forward it to peers. |
| 561 | logger.debug('advertising inv with hash: %s', hexlify(inventoryHash)) |
| 562 | broadcastToSendDataQueues((streamNumber, 'advertiseobject', inventoryHash)) |
| 563 | |
| 564 | # Now let's queue it to be processed ourselves. |
| 565 | objectProcessorQueue.put((objectType,data)) |
| 566 | |
| 567 | # If you want to command all of the sendDataThreads to do something, like shutdown or send some data, this |
| 568 | # function puts your data into the queues for each of the sendDataThreads. The sendDataThreads are |
no test coverage detected