Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
M
mybee
Project
Project
Details
Activity
Releases
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
vicotor
mybee
Commits
0b7f1c63
Unverified
Commit
0b7f1c63
authored
Jun 05, 2020
by
Zahoor Mohamed
Committed by
GitHub
Jun 05, 2020
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Assume that tags cannot be present for all chunks since bzz api didnt implement it (#260)
parent
c1bac941
Changes
2
Hide whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
56 additions
and
6 deletions
+56
-6
pusher.go
pkg/pusher/pusher.go
+9
-4
pusher_test.go
pkg/pusher/pusher_test.go
+47
-2
No files found.
pkg/pusher/pusher.go
View file @
0b7f1c63
...
...
@@ -86,9 +86,11 @@ func (s *Service) chunksWorker() {
t
,
err
:=
s
.
tag
.
GetByAddress
(
ch
.
Address
())
if
err
!=
nil
{
s
.
logger
.
Debugf
(
"pusher: get tag by address %s: %v"
,
ch
.
Address
(),
err
)
continue
//continue // // until bzz api implements tags dont continue here
}
else
{
// update the tags only if we get it
t
.
Inc
(
tags
.
StateSent
)
}
t
.
Inc
(
tags
.
StateSent
)
// Later when we process receipt, get the receipt and process it
// for now ignoring the receipt and checking only for error
...
...
@@ -137,9 +139,12 @@ func (s *Service) setChunkAsSynced(ctx context.Context, addr swarm.Address) {
ta
,
err
:=
s
.
tag
.
GetByAddress
(
addr
)
if
err
!=
nil
{
s
.
logger
.
Debugf
(
"pusher: get tag by address %s: %v"
,
addr
,
err
)
return
// return // until bzz api implements tags dont retunrn here
}
else
{
// update the tags only if we get it
ta
.
Inc
(
tags
.
StateSynced
)
}
ta
.
Inc
(
tags
.
StateSynced
)
}
}
...
...
pkg/pusher/pusher_test.go
View file @
0b7f1c63
...
...
@@ -7,12 +7,13 @@ package pusher_test
import
(
"context"
"errors"
"github.com/ethersphere/bee/pkg/tags"
"io/ioutil"
"sync"
"testing"
"time"
"github.com/ethersphere/bee/pkg/tags"
"github.com/ethersphere/bee/pkg/localstore"
"github.com/ethersphere/bee/pkg/logging"
"github.com/ethersphere/bee/pkg/pusher"
...
...
@@ -89,6 +90,50 @@ func TestSendChunkToPushSync(t *testing.T) {
p
.
Close
()
}
// TestSendChunkToPushSyncWithoutTag is similar to TestSendChunkToPushSync, excep that the tags are not
// present to simulate bzz api withotu splitter condition
func
TestSendChunkToPushSyncWithoutTag
(
t
*
testing
.
T
)
{
chunk
:=
createChunk
()
// create a trigger and a closestpeer
triggerPeer
:=
swarm
.
MustParseHexAddress
(
"6000000000000000000000000000000000000000000000000000000000000000"
)
closestPeer
:=
swarm
.
MustParseHexAddress
(
"f000000000000000000000000000000000000000000000000000000000000000"
)
pushSyncService
:=
pushsyncmock
.
New
(
func
(
ctx
context
.
Context
,
chunk
swarm
.
Chunk
)
(
*
pushsync
.
Receipt
,
error
)
{
receipt
:=
&
pushsync
.
Receipt
{
Address
:
swarm
.
NewAddress
(
chunk
.
Address
()
.
Bytes
()),
}
return
receipt
,
nil
})
mtag
:=
tags
.
NewTags
()
_
,
err
:=
mtag
.
Create
(
"name"
,
1
,
false
)
if
err
!=
nil
{
t
.
Fatal
(
err
)
}
p
,
storer
:=
createPusher
(
t
,
triggerPeer
,
pushSyncService
,
mtag
,
mock
.
WithClosestPeer
(
closestPeer
))
defer
storer
.
Close
()
_
,
err
=
storer
.
Put
(
context
.
Background
(),
storage
.
ModePutUpload
,
chunk
)
if
err
!=
nil
{
t
.
Fatal
(
err
)
}
// Check is the chunk is set as synced in the DB.
for
i
:=
0
;
i
<
noOfRetries
;
i
++
{
// Give some time for chunk to be pushed and receipt to be received
time
.
Sleep
(
10
*
time
.
Millisecond
)
err
=
checkIfModeSet
(
chunk
.
Address
(),
storage
.
ModeSetSyncPush
,
storer
)
if
err
==
nil
{
break
}
}
if
err
!=
nil
{
t
.
Fatal
(
err
)
}
p
.
Close
()
}
// TestSendChunkAndReceiveInvalidReceipt sends a chunk to pushsync to be sent ot its closest peer and
// get a invalid receipt (not with the address of the chunk sent). The test makes sure that this error
// is received and the ModeSetSyncPush is not set for the chunk.
...
...
@@ -158,6 +203,7 @@ func TestSendChunkAndTimeoutinReceivingReceipt(t *testing.T) {
tag
.
Address
=
chunk
.
Address
()
p
,
storer
:=
createPusher
(
t
,
triggerPeer
,
pushSyncService
,
mtag
,
mock
.
WithClosestPeer
(
closestPeer
))
defer
storer
.
Close
()
defer
p
.
Close
()
_
,
err
=
storer
.
Put
(
context
.
Background
(),
storage
.
ModePutUpload
,
chunk
)
if
err
!=
nil
{
...
...
@@ -177,7 +223,6 @@ func TestSendChunkAndTimeoutinReceivingReceipt(t *testing.T) {
if
err
==
nil
{
t
.
Fatalf
(
"chunk not syned error expected"
)
}
p
.
Close
()
}
func
createChunk
()
swarm
.
Chunk
{
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment